harden(gui): centraliser fail-close repli + garde-fou logging + doc mutex (revue finale Plan 1a)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Pseudonymisation_Gui_V6.py

@@ -12,6 +12,11 @@ Le mode ``--self-test`` vérifie que tout le socle GUI V6 s'importe correctement
 from __future__ import annotations
 
 import sys
+import os
+
+# Frozen Windows : désactiver le manager ONNX legacy AVANT tout import du cœur,
+# pour éviter « cannot load module more than once per process » (hotfix CLI 6c6f653).
+os.environ.setdefault("ANON_SKIP_LEGACY_ONNX_MANAGER", "1")
 
 
 def _self_test() -> int:
@@ -52,10 +57,32 @@ def main(argv=None) -> int:
     if "--self-test" in argv:
         return _self_test()
 
-    from gui_v6.app import AnonymisationApp
+    try:
+        from gui_v6.logging_setup import setup_file_logging
 
-    application = AnonymisationApp()
-    application.mainloop()
+        setup_file_logging()
+    except Exception:
+        pass
+
+    from gui_v6.app import AnonymisationApp
+    from gui_v6.single_instance import AlreadyRunningError, SingleInstance
+
+    guard = SingleInstance()
+    try:
+        guard.acquire()
+    except AlreadyRunningError:
+        try:
+            import tkinter.messagebox as mb
+
+            mb.showinfo("Anonymisation", "L'application est déjà ouverte.")
+        except Exception:
+            print("L'application est déjà ouverte.")
+        return 0
+    try:
+        application = AnonymisationApp()
+        application.mainloop()
+    finally:
+        guard.release()
     return 0
 
 

gui_v6/app.py

@@ -11,6 +11,7 @@ La fenêtre n'est créée qu'à l'instanciation de :class:`AnonymisationApp`.
 
 from __future__ import annotations
 
+import os
 from typing import Optional
 
 import customtkinter as ctk
@@ -19,10 +20,39 @@ from gui_v6 import theme as theme_mod
 from gui_v6 import ui_kit
 from gui_v6.config_state import ConfigState
 from gui_v6.license_client import LicenseClient, LicenseStatus
+from gui_v6.machine_id import default_machine_id
 from gui_v6.tabs.tab_about import AboutTab
 from gui_v6.tabs.tab_config import ConfigTab
 from gui_v6.tabs.tab_usage import UsageTab
 
+DEFAULT_PORTAL_URL = "https://app.aivanov.eu"
+
+
+def resolve_portal_url() -> str:
+    """URL du portail : env ``ANON_PORTAL_URL`` sinon défaut prod."""
+    return os.environ.get("ANON_PORTAL_URL", DEFAULT_PORTAL_URL)
+
+
+def bound_local_status(status: LicenseStatus, local_machine_id: str) -> LicenseStatus:
+    """Annoter le statut licence selon le binding poste.
+
+    Souple (décision D1) : on N'EMPÊCHE PAS le traitement. Si la licence locale
+    est valide mais liée à un autre ``machine_id`` que le poste courant (ex.
+    ``license.json`` copié), on le **signale** par un statut non valide d'affichage.
+    """
+    if status.valid and status.machine_id and status.machine_id != local_machine_id:
+        return LicenseStatus(
+            valid=False,
+            status="autre_poste",
+            message="Licence liée à un autre poste",
+            expires_at=status.expires_at,
+            grace_days=status.grace_days,
+            machine_id=status.machine_id,
+            license_ref=status.license_ref,
+        )
+    return status
+
+
 _TABS = [
     ("use", "📄 Utilisation"),
     ("cfg", "⚙️ Administration"),
@@ -38,7 +68,7 @@ class AnonymisationApp(ctk.CTk):
     ) -> None:
         super().__init__()
         self._theme_name = theme_name
-        self._license_client = license_client or LicenseClient("http://localhost")
+        self._license_client = license_client or LicenseClient(resolve_portal_url())
         self._config = ConfigState()
         self._active = "use"
         self._tab_buttons: dict = {}
@@ -80,7 +110,8 @@ class AnonymisationApp(ctk.CTk):
 
     def _safe_local_status(self) -> LicenseStatus:
         try:
-            return self._license_client.local_status()
+            status = self._license_client.local_status()
+            return bound_local_status(status, default_machine_id())
         except Exception:
             return LicenseStatus.unavailable()
 
@@ -196,7 +227,7 @@ class AnonymisationApp(ctk.CTk):
             if session is None:
                 return
             status = self._safe_local_status()
-            base_url = getattr(self._license_client, "_base_url", "") or "http://localhost"
+            base_url = getattr(self._license_client, "_base_url", "") or resolve_portal_url()
             usage_telemetry.report_run_summary(
                 summary,
                 base_url=base_url,

gui_v6/engine_bridge.py

@@ -41,6 +41,14 @@ class ManagerState(str, Enum):
     UNAVAILABLE = "unavailable"
 
 
+class EngineUnavailableError(RuntimeError):
+    """Levée quand un moteur de détection OBLIGATOIRE n'a pas pu être chargé.
+
+    Garantit le fail-close : on refuse de produire une sortie plutôt que de
+    livrer un document potentiellement non anonymisé (aligné sur le code 3 du CLI).
+    """
+
+
 @dataclass
 class EngineSettings:
     """Réglages d'appel moteur exposés par l'onglet Configuration."""
@@ -216,12 +224,21 @@ def make_process_fn(
 
     ``engine`` est injectable pour les tests ; par défaut, import paresseux de
     ``process_document`` (aucun chargement du moteur à l'import de ce module).
+
+    Raises:
+        EngineUnavailableError: si le NER obligatoire (CamemBERT-bio) est
+            indisponible alors que ``use_local_ner`` est actif (fail-close).
     """
     managers = managers if managers is not None else NerManagers(settings)
 
     def process_fn(doc_path: Path, out_dir: Path) -> dict:
         if settings.use_local_ner:
-            managers.ensure_loaded()
+            state = managers.ensure_loaded()
+            if state == ManagerState.UNAVAILABLE:
+                raise EngineUnavailableError(
+                    "Modèle de détection obligatoire (CamemBERT-bio) indisponible — "
+                    "traitement refusé pour éviter une anonymisation incomplète."
+                )
         kwargs = build_engine_kwargs(settings, managers)
         run = engine
         if run is None:

gui_v6/logging_setup.py

@@ -0,0 +1,67 @@
+"""Configuration du log fichier de la GUI V6 (E1).
+
+Sans ceci, la GUI frozen fenêtrée (sans console) perd ses logs de diagnostic.
+Le log est posé dans le même répertoire applicatif que la licence
+(``%LOCALAPPDATA%/Aivanov/Anonymisation``) pour faciliter sa récupération (E2/E3).
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from logging.handlers import RotatingFileHandler
+from pathlib import Path
+
+_CONFIGURED = False
+
+
+def _app_data_dir() -> Path:
+    base = os.environ.get("LOCALAPPDATA")
+    if base:
+        root = Path(base)
+    else:  # Linux/dev
+        root = Path.home() / ".local" / "share"
+    return root / "Aivanov" / "Anonymisation"
+
+
+def log_file_path() -> Path:
+    return _app_data_dir() / "logs" / "anonymisation.log"
+
+
+def setup_file_logging() -> Path:
+    """Configure un handler fichier rotatif sur le logger racine. Idempotent."""
+    global _CONFIGURED
+    path = log_file_path()
+    if _CONFIGURED:
+        return path
+    path.parent.mkdir(parents=True, exist_ok=True)
+    handler = RotatingFileHandler(
+        str(path), maxBytes=2_000_000, backupCount=3, encoding="utf-8"
+    )
+    handler.setFormatter(
+        logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s")
+    )
+    root = logging.getLogger()
+    if root.level == logging.NOTSET:
+        root.setLevel(logging.INFO)
+    root.addHandler(handler)
+    # Best-effort : si le cœur utilise loguru, on ajoute aussi un sink fichier.
+    try:
+        from loguru import logger as _loguru
+
+        _loguru.add(str(path), rotation="2 MB", retention=3, encoding="utf-8")
+    except Exception:
+        pass
+    _CONFIGURED = True
+    return path
+
+
+def _reset_for_tests() -> None:
+    """Réinitialise l'état pour l'isolation des tests (NE PAS appeler en prod)."""
+    global _CONFIGURED
+    root = logging.getLogger()
+    for h in list(root.handlers):
+        if isinstance(h, RotatingFileHandler):
+            root.removeHandler(h)
+            h.close()
+    _CONFIGURED = False

gui_v6/processing_runner.py

@@ -130,10 +130,11 @@ class RunSummary:
 
 
 def _default_process_fn(doc_path: Path, out_dir: Path) -> dict:
-    # Import paresseux : aucun manager NER chargé à l'import du runner.
-    from anonymizer_core_refactored_onnx import process_document
+    # Passe par make_process_fn pour bénéficier du fail-close P0-1 (refus si le
+    # NER obligatoire est indisponible), même sur ce chemin de repli.
+    from gui_v6.engine_bridge import EngineSettings, make_process_fn
 
-    return process_document(doc_path, out_dir)
+    return make_process_fn(EngineSettings())(doc_path, out_dir)
 
 
 class ProcessingRunner:

gui_v6/single_instance.py

@@ -0,0 +1,71 @@
+"""Protection multi-instance de la GUI V6 (P0-7).
+
+- Windows (frozen) : mutex nommé kernel via ctypes — ce nom DEVRA être déclaré comme
+  ``AppMutex`` dans installer/Anonymisation.iss (Plan 3 / D8) pour que l'installeur
+  ferme l'app avant une mise à jour.
+- POSIX (dev/test) : verrou ``fcntl`` exclusif sur un fichier dans le dossier app.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+
+# Nom destiné à l'AppMutex de installer/Anonymisation.iss (Plan 3 / D8). NE PAS modifier sans synchroniser le .iss.
+APP_MUTEX_NAME = "AivanonymAnonymisationV6"
+
+
+class AlreadyRunningError(RuntimeError):
+    """Une autre instance de l'application est déjà en cours d'exécution."""
+
+
+def _lock_dir() -> Path:
+    base = os.environ.get("LOCALAPPDATA")
+    root = Path(base) if base else Path.home() / ".local" / "share"
+    d = root / "Aivanov" / "Anonymisation"
+    d.mkdir(parents=True, exist_ok=True)
+    return d
+
+
+class SingleInstance:
+    def __init__(self) -> None:
+        self._handle = None  # mutex Windows
+        self._fh = None       # file handle POSIX
+
+    def acquire(self) -> None:
+        if sys.platform.startswith("win"):
+            self._acquire_windows()
+        else:
+            self._acquire_posix()
+
+    def _acquire_windows(self) -> None:  # pragma: no cover (exécuté sur Windows)
+        import ctypes
+
+        ERROR_ALREADY_EXISTS = 183
+        handle = ctypes.windll.kernel32.CreateMutexW(None, False, APP_MUTEX_NAME)
+        if not handle or ctypes.windll.kernel32.GetLastError() == ERROR_ALREADY_EXISTS:
+            raise AlreadyRunningError("L'application est déjà ouverte.")
+        self._handle = handle
+
+    def _acquire_posix(self) -> None:
+        import fcntl
+
+        path = _lock_dir() / "instance.lock"
+        fh = open(path, "w")
+        try:
+            fcntl.flock(fh.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except OSError:
+            fh.close()
+            raise AlreadyRunningError("L'application est déjà ouverte.")
+        self._fh = fh
+
+    def release(self) -> None:
+        if self._handle is not None:  # pragma: no cover
+            import ctypes
+
+            ctypes.windll.kernel32.CloseHandle(self._handle)
+            self._handle = None
+        if self._fh is not None:
+            self._fh.close()
+            self._fh = None

gui_v6/tabs/tab_about.py

@@ -33,6 +33,7 @@ _STATUS_LABELS = {
     "revoked": "Poste révoqué",
     "invalid": "Licence invalide",
     "unavailable": "Serveur de licence indisponible",
+    "autre_poste": "Licence liée à un autre poste",
     "none": "Aucune licence",
 }
 

gui_v6/theme.py

@@ -113,6 +113,7 @@ _STATUS_TOKEN = {
     "revoked": "danger",
     "invalid": "danger",
     "unavailable": "warning",
+    "autre_poste": "warning",
     "none": "text_muted",
 }
 

tests/unit/test_gui_v6_engine_failclose.py

@@ -0,0 +1,63 @@
+"""Tests fail-close P0-1 : le moteur ne doit JAMAIS être invoqué si CamemBERT-bio
+est indisponible et que use_local_ner=True.
+
+Garantie de sécurité : EngineUnavailableError est levée AVANT l'appel du moteur,
+ce qui empêche la production d'un document potentiellement non anonymisé.
+"""
+
+from pathlib import Path
+
+import pytest
+
+from gui_v6.engine_bridge import (
+    EngineSettings,
+    EngineUnavailableError,
+    NerManagers,
+    make_process_fn,
+)
+
+
+def _managers_with_broken_camembert(settings):
+    def boom():
+        raise RuntimeError("model.onnx absent")
+
+    return NerManagers(
+        settings,
+        factories={"camembert": boom, "eds": boom, "gliner": boom},
+        caps_provider=lambda: {},
+    )
+
+
+def test_process_fn_raises_when_mandatory_ner_unavailable():
+    settings = EngineSettings(use_local_ner=True)
+    managers = _managers_with_broken_camembert(settings)
+    called = {"engine": False}
+
+    def fake_engine(*a, **k):
+        called["engine"] = True
+        return {"pdf": "x"}
+
+    fn = make_process_fn(settings, managers=managers, engine=fake_engine)
+    with pytest.raises(EngineUnavailableError):
+        fn(Path("doc.pdf"), Path("/tmp/out"))
+    assert called["engine"] is False
+
+
+def test_process_fn_runs_when_ner_ok():
+    settings = EngineSettings(use_local_ner=True)
+    managers = NerManagers(
+        settings,
+        factories={"camembert": lambda: object(), "eds": lambda: None, "gliner": lambda: None},
+        caps_provider=lambda: {},
+    )
+    fn = make_process_fn(settings, managers=managers, engine=lambda *a, **k: {"pdf": "ok"})
+    assert fn(Path("d.pdf"), Path("/tmp/out")) == {"pdf": "ok"}
+
+
+def test_process_fn_skips_ner_guard_when_local_ner_disabled():
+    # use_local_ner=False : on ne charge pas le NER et on NE bloque PAS,
+    # même si les factories échoueraient (symétrique du garde-fou fail-close).
+    settings = EngineSettings(use_local_ner=False)
+    managers = _managers_with_broken_camembert(settings)
+    fn = make_process_fn(settings, managers=managers, engine=lambda *a, **k: {"pdf": "ok"})
+    assert fn(Path("d.pdf"), Path("/tmp/out")) == {"pdf": "ok"}

tests/unit/test_gui_v6_entry_frozen_flag.py

@@ -0,0 +1,16 @@
+import os
+import importlib
+
+
+def test_entry_sets_legacy_onnx_flag_on_import(monkeypatch):
+    monkeypatch.delenv("ANON_SKIP_LEGACY_ONNX_MANAGER", raising=False)
+    import Pseudonymisation_Gui_V6 as entry
+    importlib.reload(entry)
+    assert os.environ.get("ANON_SKIP_LEGACY_ONNX_MANAGER") == "1"
+
+
+def test_entry_does_not_override_explicit_flag(monkeypatch):
+    monkeypatch.setenv("ANON_SKIP_LEGACY_ONNX_MANAGER", "0")
+    import Pseudonymisation_Gui_V6 as entry
+    importlib.reload(entry)
+    assert os.environ.get("ANON_SKIP_LEGACY_ONNX_MANAGER") == "0"

tests/unit/test_gui_v6_license_binding.py

@@ -0,0 +1,29 @@
+from gui_v6.app import bound_local_status
+from gui_v6.license_client import LicenseStatus
+
+
+def test_binding_flags_other_machine():
+    st = LicenseStatus(valid=True, status="active", machine_id="AAAA1111")
+    out = bound_local_status(st, "BBBB2222")
+    assert out.valid is False
+    assert out.status == "autre_poste"
+    assert "autre poste" in out.message.lower()
+
+
+def test_binding_ok_same_machine():
+    st = LicenseStatus(valid=True, status="active", machine_id="AAAA1111")
+    out = bound_local_status(st, "AAAA1111")
+    assert out.valid is True
+    assert out.status == "active"
+
+
+def test_binding_noop_without_machine_id():
+    st = LicenseStatus(valid=True, status="active", machine_id=None)
+    assert bound_local_status(st, "AAAA1111").valid is True
+
+
+def test_binding_passes_through_invalid_status():
+    st = LicenseStatus(valid=False, status="expired", machine_id="OTHER")
+    out = bound_local_status(st, "AAAA1111")
+    assert out.status == "expired"
+    assert out.valid is False

tests/unit/test_gui_v6_logging_setup.py

@@ -0,0 +1,16 @@
+import logging
+
+
+def test_setup_file_logging_writes_to_known_path(tmp_path, monkeypatch):
+    monkeypatch.setenv("LOCALAPPDATA", str(tmp_path))
+    from gui_v6.logging_setup import setup_file_logging, _reset_for_tests
+
+    try:
+        log_path = setup_file_logging()
+        assert log_path.parent.exists()
+        logging.getLogger("test.e1").warning("ligne-temoin-42")
+        for h in logging.getLogger().handlers:
+            h.flush()
+        assert "ligne-temoin-42" in log_path.read_text(encoding="utf-8")
+    finally:
+        _reset_for_tests()

tests/unit/test_gui_v6_portal_url.py

@@ -0,0 +1,14 @@
+import pytest
+
+
+def test_default_portal_url_is_prod(monkeypatch):
+    monkeypatch.delenv("ANON_PORTAL_URL", raising=False)
+    from gui_v6.app import DEFAULT_PORTAL_URL, resolve_portal_url
+    assert DEFAULT_PORTAL_URL == "https://app.aivanov.eu"
+    assert resolve_portal_url() == "https://app.aivanov.eu"
+
+
+def test_portal_url_env_override(monkeypatch):
+    monkeypatch.setenv("ANON_PORTAL_URL", "http://localhost:8088")
+    from gui_v6.app import resolve_portal_url
+    assert resolve_portal_url() == "http://localhost:8088"

tests/unit/test_gui_v6_single_instance.py

@@ -0,0 +1,33 @@
+import pytest
+
+from gui_v6.single_instance import (
+    APP_MUTEX_NAME,
+    AlreadyRunningError,
+    SingleInstance,
+)
+
+
+def test_mutex_name_is_stable():
+    # Nom destiné à l'installeur (Inno AppMutex, Plan 3). Sentinelle anti-renommage accidentel.
+    assert APP_MUTEX_NAME == "AivanonymAnonymisationV6"
+
+
+def test_second_instance_is_rejected(tmp_path, monkeypatch):
+    monkeypatch.setenv("LOCALAPPDATA", str(tmp_path))
+    first = SingleInstance()
+    first.acquire()
+    try:
+        with pytest.raises(AlreadyRunningError):
+            SingleInstance().acquire()
+    finally:
+        first.release()
+
+
+def test_release_allows_reacquire(tmp_path, monkeypatch):
+    monkeypatch.setenv("LOCALAPPDATA", str(tmp_path))
+    a = SingleInstance()
+    a.acquire()
+    a.release()
+    b = SingleInstance()
+    b.acquire()  # ne lève pas
+    b.release()