snoopy/pyssh-client
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Assistant IA optionnel (analyse de sortie, diagnostic, generation de commande)

Browse Source 
Nouvelle couche app/ai/ (provider OpenAI-compatible via urllib, redaction de
secrets, detection de commandes dangereuses, parametres, worker QThread) et son
integration dans chaque onglet terminal via un bouton "Assistant IA" (memes
conventions que le bouton Snippets existant) :

- Expliquer cette sortie / Diagnostiquer cette erreur / Proposer une solution /
  Poser une question... : utilisent la selection courante du terminal, affichees
  dans un panneau lateral (memoire uniquement, jamais journalise) reutilisant le
  splitter comme le panneau SFTP.
- Generer une commande : ne s'execute jamais automatiquement (Copier / Inserer
  dans le terminal sans Entree / Annuler), avec avertissement si la commande
  proposee correspond a un motif destructeur connu.
- Avant tout envoi : avis unique (persist) puis detection de secrets evidents
  (mot de passe, cle, jeton) avec choix de masquer, envoyer tel quel ou annuler.

Desactive par defaut, sans liste de fournisseur figee (URL + modele libres,
compatible OpenAI/OpenRouter/LLM local). La cle API passe exclusivement par
CredentialStore (jamais dans settings.json ni journalisee). Parametrable via
le nouveau menu "Assistant IA > Parametres..." (bouton "Tester la connexion").

Bug trouve par les tests avant commit : un panneau nouvellement insere dans un
QSplitter deja visible est affiche automatiquement par Qt, ce qui inversait le
premier clic sur "Afficher le panneau IA" ; corrige en le masquant explicitement
a sa creation.

9 nouveaux fichiers de test (provider, redaction, dangers, parametres, worker,
confirmations de securite, dialogue de commande, panneau, integration onglet) :
39 verifications, toutes au vert. Suite complete (31 fichiers) executee sans
regression.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Florian
2026-09-05 05:48:13 +02:00
parent 9216f636bd
commit 0b2ff38bfd
22 changed files with 2081 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
app/ai/__init__.py Normal file
View File

37
app/ai/danger.py Normal file
View File

@@ -0,0 +1,37 @@
"""Detection locale simple de commandes potentiellement destructives, proposees
par l'assistant IA. N'a pas besoin d'etre parfaite : sert a afficher un
avertissement, jamais a bloquer ou executer quoi que ce soit automatiquement."""
from __future__ import annotations
import re
from dataclasses import dataclass
_DANGEROUS_PATTERNS: list[tuple[str, re.Pattern]] = [
("rm -rf (suppression recursive forcee)", re.compile(r"\brm\s+(-\w*\s+)*-\w*[rf]\w*[rf]\w*\b", re.IGNORECASE)),
("mkfs (formatage)", re.compile(r"\bmkfs(\.\w+)?\b", re.IGNORECASE)),
("dd (ecriture bas niveau sur un disque)", re.compile(r"\bdd\s+.*\bif=", re.IGNORECASE | re.DOTALL)),
("wipefs (effacement des signatures de systeme de fichiers)", re.compile(r"\bwipefs\b", re.IGNORECASE)),
("shutdown (extinction)", re.compile(r"\bshutdown\b", re.IGNORECASE)),
("reboot (redemarrage)", re.compile(r"\breboot\b", re.IGNORECASE)),
("iptables : purge des regles", re.compile(r"\biptables\b.*(-F|--flush)\b", re.IGNORECASE)),
("nft : purge des regles", re.compile(r"\bnft\b.*\bflush\b", re.IGNORECASE)),
("DROP DATABASE", re.compile(r"\bDROP\s+DATABASE\b", re.IGNORECASE)),
("DELETE FROM sans WHERE", re.compile(r"\bDELETE\s+FROM\s+\S+\s*;", re.IGNORECASE)),
("docker system prune (suppression en masse)", re.compile(r"\bdocker\s+system\s+prune\b", re.IGNORECASE)),
("zpool destroy", re.compile(r"\bzpool\s+destroy\b", re.IGNORECASE)),
("lvremove (suppression de volume logique)", re.compile(r"\blvremove\b", re.IGNORECASE)),
("vgremove (suppression de groupe de volumes)", re.compile(r"\bvgremove\b", re.IGNORECASE)),
]
@dataclass
class DangerMatch:
label: str
def check_dangerous(command: str) -> list[DangerMatch]:
return [DangerMatch(label) for label, pattern in _DANGEROUS_PATTERNS if pattern.search(command)]
def is_dangerous(command: str) -> bool:
return bool(check_dangerous(command))

154
app/ai/provider.py Normal file
View File

@@ -0,0 +1,154 @@
"""Abstraction du fournisseur IA : une API compatible OpenAI (endpoint chat
completions), avec URL de base et modele configurables - pas de liste de
fournisseurs ou de modeles figee en dur, pour supporter OpenAI, OpenRouter, un
proxy interne ou un LLM local exposant la meme API.
AIProvider porte les 4 usages metier (explain_output/diagnose/generate_command/
ask), tous construits au-dessus de chat(), seule methode que doit implementer
un fournisseur concret (OpenAICompatibleProvider)."""
from __future__ import annotations
import abc
import json
import urllib.error
import urllib.request
class AIError(Exception):
"""Message deja comprehensible, pret a etre affiche a l'utilisateur."""
_SYSTEM_PROMPT_TROUBLESHOOT = """Tu es un assistant specialise en administration systeme, Linux, Windows,
reseau, SSH, Docker, virtualisation et troubleshooting.
Analyse la sortie terminal fournie par l'utilisateur.
Explique :
1. ce que signifie la sortie ;
2. la cause probable ;
3. les verifications recommandees ;
4. les commandes eventuellement utiles.
Ne suppose jamais qu'une commande a ete executee.
Ne pretends jamais connaitre des elements qui ne sont pas presents.
Signale explicitement les hypotheses."""
_SYSTEM_PROMPT_DIAGNOSE = _SYSTEM_PROMPT_TROUBLESHOOT + """
Structure ta reponse ainsi :
Cause probable
Explication
Verifications
Commandes proposees
Niveau de risque"""
_SYSTEM_PROMPT_COMMAND = """Tu es un assistant specialise en administration systeme, Linux, Windows,
reseau, SSH, Docker et virtualisation.
L'utilisateur decrit ce qu'il veut accomplir en langage naturel. Propose UNE
commande (ou une courte suite de commandes) qui repond a la demande, dans un
bloc de code, avec une breve explication de ce qu'elle fait.
Ne suppose jamais le systeme d'exploitation ou l'environnement si l'utilisateur
ne l'a pas precise : demande-le, ou propose une variante par systeme si
pertinent. La commande proposee ne sera jamais executee automatiquement : elle
sera seulement affichee a l'utilisateur pour verification."""
class AIProvider(abc.ABC):
@abc.abstractmethod
def chat(self, messages: list[dict], *, temperature: float | None = None) -> str:
"""Renvoie le texte de la reponse. Leve AIError en cas d'echec."""
def explain_output(self, output: str) -> str:
return self.chat([
{"role": "system", "content": _SYSTEM_PROMPT_TROUBLESHOOT},
{"role": "user", "content": f"Explique cette sortie terminal :\n\n{output}"},
])
def diagnose(self, output: str) -> str:
return self.chat([
{"role": "system", "content": _SYSTEM_PROMPT_DIAGNOSE},
{"role": "user", "content": f"Diagnostique cette erreur :\n\n{output}"},
])
def propose_solution(self, output: str) -> str:
return self.chat([
{"role": "system", "content": _SYSTEM_PROMPT_TROUBLESHOOT},
{"role": "user", "content": (
"Propose une solution concrete a ce probleme, avec les commandes "
f"eventuellement necessaires :\n\n{output}"
)},
])
def generate_command(self, request: str) -> str:
return self.chat([
{"role": "system", "content": _SYSTEM_PROMPT_COMMAND},
{"role": "user", "content": request},
])
def ask(self, question: str, context: str = "") -> str:
messages = [{"role": "system", "content": _SYSTEM_PROMPT_TROUBLESHOOT}]
if context:
messages.append({"role": "user", "content": f"Contexte (sortie terminal selectionnee) :\n{context}"})
messages.append({"role": "user", "content": question})
return self.chat(messages)
class OpenAICompatibleProvider(AIProvider):
def __init__(
self, base_url: str, api_key: str, model: str,
timeout: float = 30.0, temperature: float = 0.3,
):
self.base_url = base_url.rstrip("/")
self.api_key = api_key
self.model = model
self.timeout = timeout
self.temperature = temperature
def chat(self, messages: list[dict], *, temperature: float | None = None) -> str:
if not self.base_url:
raise AIError("Aucune URL de base configuree pour l'assistant IA.")
if not self.model:
raise AIError("Aucun modele configure pour l'assistant IA.")
url = f"{self.base_url}/chat/completions"
payload = {
"model": self.model,
"messages": messages,
"temperature": self.temperature if temperature is None else temperature,
}
headers = {"Content-Type": "application/json"}
if self.api_key:
headers["Authorization"] = f"Bearer {self.api_key}"
request = urllib.request.Request(
url, data=json.dumps(payload).encode("utf-8"), headers=headers, method="POST",
)
try:
with urllib.request.urlopen(request, timeout=self.timeout) as response:
raw = response.read()
except urllib.error.HTTPError as exc:
raise AIError(self._describe_http_error(exc)) from exc
except urllib.error.URLError as exc:
if isinstance(exc.reason, TimeoutError):
raise AIError("Le fournisseur IA n'a pas repondu a temps (delai depasse).") from exc
raise AIError(f"Impossible de joindre le fournisseur IA : {exc.reason}") from exc
try:
parsed = json.loads(raw.decode("utf-8"))
return parsed["choices"][0]["message"]["content"]
except (json.JSONDecodeError, KeyError, IndexError, TypeError, UnicodeDecodeError) as exc:
raise AIError(f"Reponse du fournisseur IA invalide ou inattendue : {exc}") from exc
@staticmethod
def _describe_http_error(exc: urllib.error.HTTPError) -> str:
if exc.code == 401:
return "Authentification refusee (401) : verifiez la cle API."
if exc.code == 429:
return "Trop de requetes (429) : le fournisseur limite le debit, reessayez plus tard."
try:
body = exc.read().decode("utf-8", errors="replace")
except Exception: # noqa: BLE001
body = ""
return f"Erreur du fournisseur IA ({exc.code}) : {(body or exc.reason)[:300]}"

52
app/ai/redaction.py Normal file
View File

@@ -0,0 +1,52 @@
"""Detection/redaction basique de secrets avant envoi a un fournisseur IA
externe : cles privees SSH, jetons Bearer, cles AWS, en-tetes Authorization,
password=/token=/api_key=, et longues chaines ressemblant a un secret.
Heuristique volontairement simple (comme la detection de commandes
dangereuses) : mieux vaut avertir a tort sur un hash ou un identifiant anodin
que laisser passer un vrai secret."""
from __future__ import annotations
import re
from dataclasses import dataclass
_PATTERNS: list[tuple[str, re.Pattern]] = [
("Cle privee (PEM/OpenSSH)", re.compile(
r"-----BEGIN (?:RSA |EC |OPENSSH |DSA |ENCRYPTED )?PRIVATE KEY-----"
)),
("Jeton Bearer", re.compile(r"\bBearer\s+[A-Za-z0-9\-_.=]{10,}", re.IGNORECASE)),
("Cle d'acces AWS", re.compile(r"\b(?:AKIA|ASIA)[0-9A-Z]{16}\b")),
("En-tete Authorization", re.compile(r"\bAuthorization\s*:\s*\S+", re.IGNORECASE)),
("Mot de passe (password=)", re.compile(r"\b(?:password|passwd|pwd)\s*[:=]\s*\S+", re.IGNORECASE)),
("Cle API (api_key=)", re.compile(r"\b(?:api[_-]?key|apikey)\s*[:=]\s*\S+", re.IGNORECASE)),
("Jeton (token=)", re.compile(r"\btoken\s*[:=]\s*\S+", re.IGNORECASE)),
("Chaine longue ressemblant a un secret", re.compile(r"\b[A-Za-z0-9_\-]{32,}\b")),
]
@dataclass
class DetectedSecret:
label: str
excerpt: str
def detect_secrets(text: str) -> list[DetectedSecret]:
"""Renvoie les categories detectees (au plus une occurrence par categorie,
pour un resume lisible), dans l'ordre des motifs (du plus specifique au
plus generique)."""
found: list[DetectedSecret] = []
for label, pattern in _PATTERNS:
match = pattern.search(text)
if match:
excerpt = match.group(0)
if len(excerpt) > 60:
excerpt = excerpt[:57] + "..."
found.append(DetectedSecret(label, excerpt))
return found
def redact(text: str) -> str:
redacted = text
for _label, pattern in _PATTERNS:
redacted = pattern.sub("[MASQUE]", redacted)
return redacted

54
app/ai/service.py Normal file
View File

@@ -0,0 +1,54 @@
"""Execution des requetes IA en arriere-plan : aucun appel reseau ne doit
jamais bloquer le thread Qt."""
from __future__ import annotations
from PySide6.QtCore import QThread, Signal
from app.ai.provider import AIProvider, AIError
KIND_EXPLAIN = "explain"
KIND_DIAGNOSE = "diagnose"
KIND_SOLUTION = "solution"
KIND_GENERATE = "generate"
KIND_ASK = "ask"
KIND_TEST = "test"
class AIRequestWorker(QThread):
"""Une requete = un worker jetable. L'annulation est cooperative : la
requete HTTP en cours (bornee par son timeout) va a son terme, mais le
signal de resultat peut etre ignore par l'appelant (voir AIPanel)."""
finished_ok = Signal(str)
finished_error = Signal(str)
def __init__(self, provider: AIProvider, kind: str, payload: str, context: str = "", parent=None):
super().__init__(parent)
self.provider = provider
self.kind = kind
self.payload = payload
self.context = context
def run(self) -> None:
try:
result = self._execute()
self.finished_ok.emit(result)
except AIError as exc:
self.finished_error.emit(str(exc))
except Exception as exc: # noqa: BLE001 - ne jamais laisser un thread planter silencieusement
self.finished_error.emit(f"Erreur inattendue : {exc}")
def _execute(self) -> str:
if self.kind == KIND_EXPLAIN:
return self.provider.explain_output(self.payload)
if self.kind == KIND_DIAGNOSE:
return self.provider.diagnose(self.payload)
if self.kind == KIND_SOLUTION:
return self.provider.propose_solution(self.payload)
if self.kind == KIND_GENERATE:
return self.provider.generate_command(self.payload)
if self.kind == KIND_ASK:
return self.provider.ask(self.payload, context=self.context)
if self.kind == KIND_TEST:
return self.provider.chat([{"role": "user", "content": "Reponds uniquement 'OK'."}])
raise AIError(f"Type de requete IA inconnu : {self.kind}")

85
app/ai/settings.py Normal file
View File

@@ -0,0 +1,85 @@
"""Configuration de l'assistant IA (optionnel, desactive par defaut).
La cle API n'est jamais stockee ici : seuls les parametres non sensibles
(activation, fournisseur, URL, modele, timeout, temperature) le sont, via
app.app_settings. La cle passe exclusivement par CredentialStore
(app.credentials), jamais dans settings.json/sessions.json/logs."""
from __future__ import annotations
from dataclasses import dataclass
from app.app_settings import get_setting, set_setting
from app.credentials import CredentialStore
from app.ai.provider import AIProvider, OpenAICompatibleProvider
AI_API_KEY_CREDENTIAL = "ai:api_key"
DEFAULT_BASE_URL = "https://api.openai.com/v1"
DEFAULT_PROVIDER_NAME = "OpenAI"
DEFAULT_MODEL = ""
DEFAULT_TIMEOUT = 30.0
DEFAULT_TEMPERATURE = 0.3
@dataclass
class AISettings:
enabled: bool = False
provider_name: str = DEFAULT_PROVIDER_NAME
base_url: str = DEFAULT_BASE_URL
model: str = DEFAULT_MODEL
timeout: float = DEFAULT_TIMEOUT
temperature: float = DEFAULT_TEMPERATURE
def load_ai_settings() -> AISettings:
try:
timeout = float(get_setting("ai_timeout", DEFAULT_TIMEOUT))
except (TypeError, ValueError):
timeout = DEFAULT_TIMEOUT
try:
temperature = float(get_setting("ai_temperature", DEFAULT_TEMPERATURE))
except (TypeError, ValueError):
temperature = DEFAULT_TEMPERATURE
return AISettings(
enabled=bool(get_setting("ai_enabled", False)),
provider_name=get_setting("ai_provider_name", DEFAULT_PROVIDER_NAME) or DEFAULT_PROVIDER_NAME,
base_url=get_setting("ai_base_url", DEFAULT_BASE_URL) or DEFAULT_BASE_URL,
model=get_setting("ai_model", DEFAULT_MODEL) or DEFAULT_MODEL,
timeout=timeout,
temperature=temperature,
)
def save_ai_settings(settings: AISettings) -> None:
set_setting("ai_enabled", settings.enabled)
set_setting("ai_provider_name", settings.provider_name)
set_setting("ai_base_url", settings.base_url)
set_setting("ai_model", settings.model)
set_setting("ai_timeout", settings.timeout)
set_setting("ai_temperature", settings.temperature)
def load_api_key(credential_store: CredentialStore) -> str:
return credential_store.get_secret(AI_API_KEY_CREDENTIAL) or ""
def save_api_key(credential_store: CredentialStore, api_key: str) -> bool:
if api_key:
return credential_store.set_secret(AI_API_KEY_CREDENTIAL, api_key)
credential_store.delete_secret(AI_API_KEY_CREDENTIAL)
return True
def build_provider_from_settings(credential_store: CredentialStore) -> AIProvider | None:
"""None si l'assistant est desactive : point d'appel unique pour tous les
endroits de l'UI qui ont besoin d'un fournisseur pret a l'emploi."""
settings = load_ai_settings()
if not settings.enabled:
return None
return OpenAICompatibleProvider(
base_url=settings.base_url,
api_key=load_api_key(credential_store),
model=settings.model,
timeout=settings.timeout,
temperature=settings.temperature,
)

8
app/controllers/connection_controller.py
View File

@@ -33,6 +33,7 @@ class ConnectionController(QObject):
broadcast: BroadcastManager, broadcast: BroadcastManager,
snippet_store=None, snippet_store=None,
credential_store: CredentialStore | None = None, credential_store: CredentialStore | None = None,
ai_provider_factory=None,
parent_widget: QWidget | None = None, parent_widget: QWidget | None = None,
): ):
super().__init__(parent_widget) super().__init__(parent_widget)
@@ -41,6 +42,7 @@ class ConnectionController(QObject):
self.broadcast = broadcast self.broadcast = broadcast
self.snippet_store = snippet_store self.snippet_store = snippet_store
self.credential_store = credential_store or NullCredentialStore() self.credential_store = credential_store or NullCredentialStore()
self.ai_provider_factory = ai_provider_factory
self._parent_widget = parent_widget self._parent_widget = parent_widget
self.tabs = QTabWidget(parent_widget) self.tabs = QTabWidget(parent_widget)
@@ -108,6 +110,7 @@ class ConnectionController(QObject):
return SessionTabWidget( return SessionTabWidget(
session, secret, bastion_target=bastion_target, bastion_password=bastion_password, session, secret, bastion_target=bastion_target, bastion_password=bastion_password,
snippet_store=self.snippet_store, broadcast=self.broadcast, snippet_store=self.snippet_store, broadcast=self.broadcast,
ai_provider_factory=self.ai_provider_factory, credential_store=self.credential_store,
) )
def build_sftp_tab(self, session: Session) -> SFTPTabWidget | None: def build_sftp_tab(self, session: Session) -> SFTPTabWidget | None:
@@ -149,7 +152,10 @@ class ConnectionController(QObject):
def new_local_terminal(self, shell: str = DEFAULT_SHELL) -> LocalTerminalTabWidget: def new_local_terminal(self, shell: str = DEFAULT_SHELL) -> LocalTerminalTabWidget:
"""Cree un terminal local cable (multi-send, SFTP detecte), sans l'ajouter """Cree un terminal local cable (multi-send, SFTP detecte), sans l'ajouter
au QTabWidget : reutilise pour un onglet normal comme pour un panneau.""" au QTabWidget : reutilise pour un onglet normal comme pour un panneau."""
tab = LocalTerminalTabWidget(shell, snippet_store=self.snippet_store, broadcast=self.broadcast) tab = LocalTerminalTabWidget(
shell, snippet_store=self.snippet_store, broadcast=self.broadcast,
ai_provider_factory=self.ai_provider_factory, credential_store=self.credential_store,
)
tab.sftp_requested.connect(self.quick_sftp) tab.sftp_requested.connect(self.quick_sftp)
self.broadcast.register(tab) self.broadcast.register(tab)
return tab return tab

79
app/dialogs/ai_command_dialog.py Normal file
View File

@@ -0,0 +1,79 @@
"""Affiche une commande proposee par l'assistant IA : jamais executee
automatiquement, seulement copiable ou inseree dans le terminal (sans Entree)
apres verification par l'utilisateur."""
from __future__ import annotations
import re
from PySide6.QtWidgets import (
QDialog, QVBoxLayout, QHBoxLayout, QTextEdit, QPushButton, QLabel, QApplication,
)
_CODE_BLOCK_RE = re.compile(r"```(?:[a-zA-Z0-9_+-]*\n)?(.*?)```", re.DOTALL)
def extract_command(response: str) -> str:
"""Prend le contenu du premier bloc de code s'il y en a un (le prompt
systeme en demande un), sinon la reponse entiere : mieux vaut proposer un
texte un peu trop large que perdre silencieusement la commande."""
match = _CODE_BLOCK_RE.search(response)
text = match.group(1) if match else response
return text.strip()
class AICommandDialog(QDialog):
def __init__(self, response: str, dangerous_labels: list[str] | None = None, parent=None):
super().__init__(parent)
self.setWindowTitle("Commande generee par l'assistant IA")
self.setMinimumSize(520, 360)
self.command = extract_command(response)
self.insert_requested = False
layout = QVBoxLayout(self)
if dangerous_labels:
warning = QLabel(
"⚠ Commande potentiellement destructive : " + ", ".join(dangerous_labels)
)
warning.setWordWrap(True)
warning.setStyleSheet("color: #c0392b; font-weight: bold;")
layout.addWidget(warning)
layout.addWidget(QLabel("Reponse complete de l'assistant :"))
self.response_view = QTextEdit(response)
self.response_view.setReadOnly(True)
layout.addWidget(self.response_view, 1)
layout.addWidget(QLabel("Commande proposee (celle qui sera copiee/inseree) :"))
self.command_view = QTextEdit(self.command)
self.command_view.setFixedHeight(70)
layout.addWidget(self.command_view)
self.status_label = QLabel("Cette commande ne sera jamais executee automatiquement.")
self.status_label.setObjectName("statusLabel")
layout.addWidget(self.status_label)
btn_row = QHBoxLayout()
self.copy_btn = QPushButton("Copier")
self.insert_btn = QPushButton("Inserer dans le terminal")
self.cancel_btn = QPushButton("Annuler")
for b in (self.copy_btn, self.insert_btn, self.cancel_btn):
btn_row.addWidget(b)
layout.addLayout(btn_row)
self.copy_btn.clicked.connect(self._copy)
self.insert_btn.clicked.connect(self._insert)
self.cancel_btn.clicked.connect(self.reject)
def _current_command(self) -> str:
# l'utilisateur peut corriger la commande proposee avant de l'utiliser
return self.command_view.toPlainText().strip()
def _copy(self) -> None:
QApplication.clipboard().setText(self._current_command())
self.status_label.setText("Copie dans le presse-papiers.")
def _insert(self) -> None:
self.command = self._current_command()
self.insert_requested = True
self.accept()

70
app/dialogs/ai_prompts.py Normal file
View File

@@ -0,0 +1,70 @@
"""Confirmations de securite avant tout envoi de contenu au fournisseur IA :
un avertissement de principe affiche une seule fois, puis une verification
systematique des secrets evidents a chaque envoi d'une selection de terminal
ou d'une demande libre."""
from __future__ import annotations
from PySide6.QtWidgets import QMessageBox
from app.app_settings import get_setting, set_setting
from app.ai.redaction import detect_secrets
_NOTICE_ACK_KEY = "ai_notice_acknowledged"
MASK = "mask"
SEND = "send"
CANCEL = "cancel"
def ensure_ai_notice_acknowledged(parent) -> bool:
"""Affiche l'avertissement de principe une seule fois (jamais reaffiche
ensuite, l'accord etant persiste). Renvoie False si l'utilisateur refuse :
l'appelant ne doit alors rien envoyer."""
if get_setting(_NOTICE_ACK_KEY, False):
return True
box = QMessageBox(parent)
box.setWindowTitle("Assistant IA")
box.setIcon(QMessageBox.Information)
box.setText(
"Le texte selectionne dans le terminal (ou votre question) va etre envoye "
"au fournisseur IA que vous avez configure, en dehors de cet ordinateur.\n\n"
"PySSH tente de detecter les secrets evidents (mots de passe, cles, jetons) "
"avant chaque envoi et vous permet de les masquer, mais ne peut pas tout "
"detecter : verifiez ce que vous envoyez.\n\n"
"Ce message ne sera plus affiche."
)
box.setStandardButtons(QMessageBox.Ok | QMessageBox.Cancel)
box.setDefaultButton(QMessageBox.Ok)
accepted = box.exec() == QMessageBox.Ok
if accepted:
set_setting(_NOTICE_ACK_KEY, True)
return accepted
def confirm_sensitive_content(parent, text: str) -> str:
"""Renvoie MASK, SEND ou CANCEL. Si aucun secret evident n'est detecte,
renvoie SEND directement sans rien afficher."""
secrets = detect_secrets(text)
if not secrets:
return SEND
labels = "\n".join(f"- {s.label}" for s in secrets)
box = QMessageBox(parent)
box.setWindowTitle("Donnees potentiellement sensibles")
box.setIcon(QMessageBox.Warning)
box.setText(
"Des donnees potentiellement sensibles ont ete detectees dans le texte "
f"qui serait envoye au fournisseur IA :\n\n{labels}\n\n"
"« Masquer automatiquement » les remplace par [MASQUE] avant l'envoi.\n"
"« Voir le contenu » l'envoie tel quel, sans modification."
)
mask_btn = box.addButton("Masquer automatiquement", QMessageBox.AcceptRole)
send_btn = box.addButton("Voir le contenu", QMessageBox.DestructiveRole)
box.addButton("Annuler", QMessageBox.RejectRole)
box.setDefaultButton(mask_btn)
box.exec()
clicked = box.clickedButton()
if clicked is mask_btn:
return MASK
if clicked is send_btn:
return SEND
return CANCEL

161
app/dialogs/ai_settings_dialog.py Normal file
View File

@@ -0,0 +1,161 @@
"""Configuration de l'assistant IA (optionnel, desactive par defaut) : aucune
liste figee de fournisseurs/modeles, juste une URL de base et un nom de modele
compatibles avec l'API OpenAI (OpenAI, OpenRouter, proxy interne, LLM local...)."""
from __future__ import annotations
from PySide6.QtWidgets import (
QDialog, QVBoxLayout, QFormLayout, QLineEdit, QCheckBox, QDoubleSpinBox,
QPushButton, QDialogButtonBox, QMessageBox, QLabel,
)
from app.ai.settings import AISettings, load_ai_settings, save_ai_settings, load_api_key, save_api_key
from app.ai.provider import OpenAICompatibleProvider
from app.ai.service import KIND_TEST, AIRequestWorker
from app.credentials import CredentialStore, NullCredentialStore
class AISettingsDialog(QDialog):
def __init__(self, credential_store: CredentialStore | None = None, parent=None):
super().__init__(parent)
self.credential_store = credential_store or NullCredentialStore()
self._test_worker: AIRequestWorker | None = None
self.setWindowTitle("Assistant IA")
self.setMinimumWidth(460)
settings = load_ai_settings()
api_key = load_api_key(self.credential_store)
layout = QVBoxLayout(self)
form = QFormLayout()
self.enabled_check = QCheckBox("Activer l'assistant IA")
self.enabled_check.setChecked(settings.enabled)
self.enabled_check.setToolTip(
"Desactive par defaut. Aucune donnee n'est envoyee a un service externe "
"tant que cette case n'est pas cochee et qu'un modele n'est pas configure."
)
form.addRow("", self.enabled_check)
self.provider_name_edit = QLineEdit(settings.provider_name)
self.provider_name_edit.setPlaceholderText("ex : OpenAI, OpenRouter, LLM local...")
form.addRow("Fournisseur :", self.provider_name_edit)
self.base_url_edit = QLineEdit(settings.base_url)
self.base_url_edit.setPlaceholderText("https://api.openai.com/v1")
form.addRow("URL de base :", self.base_url_edit)
self.model_edit = QLineEdit(settings.model)
self.model_edit.setPlaceholderText("ex : gpt-4o-mini")
form.addRow("Modele :", self.model_edit)
self.api_key_edit = QLineEdit(api_key)
self.api_key_edit.setEchoMode(QLineEdit.Password)
self.api_key_edit.setPlaceholderText("Laisser vide si le fournisseur n'en requiert pas")
if not self.credential_store.is_available():
self.api_key_edit.setEnabled(False)
self.api_key_edit.setPlaceholderText("Aucun trousseau securise disponible sur ce systeme")
form.addRow("Cle API :", self.api_key_edit)
self.timeout_spin = QDoubleSpinBox()
self.timeout_spin.setRange(1.0, 600.0)
self.timeout_spin.setSuffix(" s")
self.timeout_spin.setValue(settings.timeout)
form.addRow("Delai d'attente :", self.timeout_spin)
self.temperature_spin = QDoubleSpinBox()
self.temperature_spin.setRange(0.0, 2.0)
self.temperature_spin.setSingleStep(0.1)
self.temperature_spin.setValue(settings.temperature)
form.addRow("Temperature :", self.temperature_spin)
layout.addLayout(form)
note = QLabel(
"La cle API est memorisee dans le trousseau securise du systeme, jamais "
"dans un fichier de configuration. L'assistant est entierement optionnel : "
"aucune fonctionnalite existante n'en depend."
)
note.setWordWrap(True)
note.setObjectName("statusLabel")
layout.addWidget(note)
test_row = QVBoxLayout()
self.test_btn = QPushButton("Tester la connexion")
self.test_btn.clicked.connect(self._test_connection)
self.test_status = QLabel("")
self.test_status.setWordWrap(True)
test_row.addWidget(self.test_btn)
test_row.addWidget(self.test_status)
layout.addLayout(test_row)
buttons = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel)
buttons.accepted.connect(self._on_accept)
buttons.rejected.connect(self.reject)
layout.addWidget(buttons)
# ---------- lecture du formulaire ----------
def _form_settings(self) -> AISettings:
return AISettings(
enabled=self.enabled_check.isChecked(),
provider_name=self.provider_name_edit.text().strip() or "OpenAI",
base_url=self.base_url_edit.text().strip(),
model=self.model_edit.text().strip(),
timeout=self.timeout_spin.value(),
temperature=self.temperature_spin.value(),
)
def _build_provider(self) -> OpenAICompatibleProvider | None:
settings = self._form_settings()
if not settings.base_url or not settings.model:
QMessageBox.warning(
self, "Configuration incomplete",
"Renseignez au moins l'URL de base et le modele avant de tester.",
)
return None
return OpenAICompatibleProvider(
base_url=settings.base_url,
api_key=self.api_key_edit.text().strip(),
model=settings.model,
timeout=settings.timeout,
temperature=settings.temperature,
)
# ---------- test de connexion ----------
def _test_connection(self) -> None:
if self._test_worker is not None and self._test_worker.isRunning():
return
provider = self._build_provider()
if provider is None:
return
self.test_btn.setEnabled(False)
self.test_status.setText("Test en cours...")
self._test_worker = AIRequestWorker(provider, KIND_TEST, "")
self._test_worker.finished_ok.connect(self._on_test_ok)
self._test_worker.finished_error.connect(self._on_test_error)
self._test_worker.start()
def _on_test_ok(self, _text: str) -> None:
self.test_btn.setEnabled(True)
self.test_status.setText("✓ Connexion reussie.")
def _on_test_error(self, message: str) -> None:
self.test_btn.setEnabled(True)
self.test_status.setText(f"✗ Echec : {message}")
# ---------- validation ----------
def _on_accept(self) -> None:
settings = self._form_settings()
if settings.enabled and (not settings.base_url or not settings.model):
QMessageBox.warning(
self, "Configuration incomplete",
"Renseignez au moins l'URL de base et le modele pour activer l'assistant IA.",
)
return
save_ai_settings(settings)
save_api_key(self.credential_store, self.api_key_edit.text().strip())
self.accept()
def closeEvent(self, event) -> None:
if self._test_worker is not None and self._test_worker.isRunning():
self._test_worker.wait(2000)
super().closeEvent(event)

15
app/main_window.py
View File

@@ -21,6 +21,8 @@ from app.dialogs.known_hosts_dialog import KnownHostsDialog
from app.dialogs.workspace_dialog import WorkspaceDialog from app.dialogs.workspace_dialog import WorkspaceDialog
from app.dialogs.snippet_dialog import SnippetManagerDialog from app.dialogs.snippet_dialog import SnippetManagerDialog
from app.dialogs.quick_connect_dialog import QuickConnectDialog from app.dialogs.quick_connect_dialog import QuickConnectDialog
from app.dialogs.ai_settings_dialog import AISettingsDialog
from app.ai.settings import build_provider_from_settings
from app.theme import THEME_LABELS, apply_theme, current_theme, save_theme from app.theme import THEME_LABELS, apply_theme, current_theme, save_theme
from app.workspace import WorkspaceStore from app.workspace import WorkspaceStore
from app.snippets import SnippetStore from app.snippets import SnippetStore
@@ -52,7 +54,9 @@ class MainWindow(QMainWindow):
self.broadcast = BroadcastManager(self._iter_tabs, self) self.broadcast = BroadcastManager(self._iter_tabs, self)
self.connection_controller = ConnectionController( self.connection_controller = ConnectionController(
self.session_store, self.key_manager, self.broadcast, self.session_store, self.key_manager, self.broadcast,
snippet_store=self.snippet_store, credential_store=self.credential_store, parent_widget=self, snippet_store=self.snippet_store, credential_store=self.credential_store,
ai_provider_factory=lambda: build_provider_from_settings(self.credential_store),
parent_widget=self,
) )
self.session_controller = SessionController( self.session_controller = SessionController(
self.session_store, self.key_manager, self.broadcast.button, parent_widget=self, self.session_store, self.key_manager, self.broadcast.button, parent_widget=self,
@@ -189,6 +193,11 @@ class MainWindow(QMainWindow):
manage_snippets_action.triggered.connect(self.open_snippet_manager) manage_snippets_action.triggered.connect(self.open_snippet_manager)
snippets_menu.addAction(manage_snippets_action) snippets_menu.addAction(manage_snippets_action)
ai_menu = self.menuBar().addMenu("&Assistant IA")
ai_settings_action = QAction("Parametres...", self)
ai_settings_action.triggered.connect(self.open_ai_settings)
ai_menu.addAction(ai_settings_action)
help_menu = self.menuBar().addMenu("&Aide") help_menu = self.menuBar().addMenu("&Aide")
about_action = QAction("A propos", self) about_action = QAction("A propos", self)
about_action.triggered.connect(self.show_about) about_action.triggered.connect(self.show_about)
@@ -246,6 +255,10 @@ class MainWindow(QMainWindow):
dialog = SnippetManagerDialog(self.snippet_store, parent=self) dialog = SnippetManagerDialog(self.snippet_store, parent=self)
dialog.exec() dialog.exec()
def open_ai_settings(self) -> None:
dialog = AISettingsDialog(credential_store=self.credential_store, parent=self)
dialog.exec()
def open_quick_connect(self) -> None: def open_quick_connect(self) -> None:
dialog = QuickConnectDialog( dialog = QuickConnectDialog(
self.session_store, self.connection_controller, self.workspace_manager, self.session_store, self.connection_controller, self.workspace_manager,

171
app/ui/ai_panel.py Normal file
View File

@@ -0,0 +1,171 @@
"""Panneau lateral de l'assistant IA : conversation en memoire uniquement
(jamais persistee, jamais journalisee), une requete a la fois.
Point d'entree unique (`run_request`) pour toutes les actions IA d'un onglet :
c'est ici, et seulement ici, que se fait la verification de securite (avis
unique + detection de secrets) avant tout envoi au fournisseur configure.
"""
from __future__ import annotations
import html
from PySide6.QtWidgets import (
QWidget, QVBoxLayout, QHBoxLayout, QTextEdit, QPushButton, QLabel, QLineEdit, QApplication,
)
from app.ai.service import AIRequestWorker, KIND_ASK
from app.ai.redaction import redact
from app.dialogs.ai_prompts import ensure_ai_notice_acknowledged, confirm_sensitive_content, MASK, CANCEL
_KIND_LABELS = {
"explain": "Expliquer cette sortie",
"diagnose": "Diagnostiquer cette erreur",
"solution": "Proposer une solution",
"ask": "Question",
}
def _escape(text: str) -> str:
return html.escape(text).replace("\n", "<br>")
class AIPanel(QWidget):
"""`provider_factory` est un callable sans argument renvoyant un AIProvider
pret a l'emploi, ou None si l'assistant est desactive/non configure -
rappele a chaque requete pour refleter immediatement un changement de
parametres, sans avoir a rouvrir l'onglet."""
def __init__(self, provider_factory, parent=None):
super().__init__(parent)
self.provider_factory = provider_factory
self._worker: AIRequestWorker | None = None
self._generation = 0
self._last_response = ""
self._pending_context = ""
layout = QVBoxLayout(self)
layout.setContentsMargins(6, 6, 6, 6)
header = QHBoxLayout()
header.addWidget(QLabel("🤖 Assistant IA"), 1)
close_btn = QPushButton("✕")
close_btn.setFixedWidth(28)
close_btn.setToolTip("Masquer le panneau")
close_btn.clicked.connect(self.hide)
header.addWidget(close_btn)
layout.addLayout(header)
self.conversation = QTextEdit()
self.conversation.setReadOnly(True)
layout.addWidget(self.conversation, 1)
self.busy_label = QLabel("")
self.busy_label.setObjectName("statusLabel")
layout.addWidget(self.busy_label)
self.question_row = QWidget()
question_layout = QHBoxLayout(self.question_row)
question_layout.setContentsMargins(0, 0, 0, 0)
self.question_edit = QLineEdit()
self.question_edit.setPlaceholderText("Votre question...")
self.question_edit.returnPressed.connect(self._send_question)
send_btn = QPushButton("Envoyer")
send_btn.clicked.connect(self._send_question)
question_layout.addWidget(self.question_edit, 1)
question_layout.addWidget(send_btn)
self.question_row.hide()
layout.addWidget(self.question_row)
btn_row = QHBoxLayout()
self.copy_btn = QPushButton("Copier reponse")
self.new_question_btn = QPushButton("Nouvelle question")
self.clear_btn = QPushButton("Effacer la conversation")
self.copy_btn.clicked.connect(self._copy_last_response)
self.new_question_btn.clicked.connect(lambda: self.prompt_question(""))
self.clear_btn.clicked.connect(self.clear_conversation)
for b in (self.copy_btn, self.new_question_btn, self.clear_btn):
btn_row.addWidget(b)
layout.addLayout(btn_row)
# ---------- conversation ----------
def clear_conversation(self) -> None:
self._generation += 1 # toute reponse en vol devient perimee et sera ignoree
self.conversation.clear()
self._last_response = ""
self.busy_label.clear()
self.question_row.hide()
def _copy_last_response(self) -> None:
if self._last_response:
QApplication.clipboard().setText(self._last_response)
# ---------- question libre ----------
def prompt_question(self, context: str = "") -> None:
"""Affiche le panneau et la zone de saisie d'une question. `context`
est un texte optionnel (ex : selection du terminal) joint a la
question, propose a la meme verification de secrets avant l'envoi."""
self.show()
self._pending_context = context
self.question_row.show()
self.question_edit.setFocus()
def _send_question(self) -> None:
question = self.question_edit.text().strip()
if not question:
return
self.question_edit.clear()
self.question_row.hide()
context, self._pending_context = self._pending_context, ""
self.run_request(KIND_ASK, question, context=context, label=_KIND_LABELS["ask"])
# ---------- requetes ----------
def run_request(self, kind: str, payload: str, context: str = "", label: str = "") -> None:
if self._worker is not None and self._worker.isRunning():
self.busy_label.setText("Une requete est deja en cours, patientez...")
return
provider = self.provider_factory() if self.provider_factory else None
if provider is None:
self.show()
self.conversation.append("<i>Assistant IA non configure (menu Assistant IA > Parametres...).</i>")
return
if not ensure_ai_notice_acknowledged(self):
return
to_check = payload + ("\n" + context if context else "")
decision = confirm_sensitive_content(self, to_check)
if decision == CANCEL:
return
if decision == MASK:
payload = redact(payload)
context = redact(context) if context else context
self.show()
label = label or _KIND_LABELS.get(kind, kind)
self.conversation.append(f"<b>{html.escape(label)}</b><br>{_escape(payload)}")
self.busy_label.setText("Interrogation du fournisseur IA...")
generation = self._generation
self._worker = AIRequestWorker(provider, kind, payload, context=context)
self._worker.finished_ok.connect(lambda text: self._on_ok(generation, text))
self._worker.finished_error.connect(lambda message: self._on_error(generation, message))
self._worker.start()
def _on_ok(self, generation: int, text: str) -> None:
if generation != self._generation:
return
self.busy_label.clear()
self._last_response = text
self.conversation.append(f"<div>{_escape(text)}</div>")
def _on_error(self, generation: int, message: str) -> None:
if generation != self._generation:
return
self.busy_label.clear()
self.conversation.append(f"<span style='color:#c0392b;'>Erreur : {_escape(message)}</span>")
def shutdown(self) -> None:
"""A appeler avant la destruction du panneau (fermeture de l'onglet) :
une requete deja en vol devient perimee (voir `_generation`) et son
thread dispose d'un court delai pour se terminer proprement."""
self._generation += 1
if self._worker is not None and self._worker.isRunning():
self._worker.wait(2000)

173
app/ui/tabs.py
View File

@@ -15,7 +15,7 @@ from PySide6.QtCore import Qt, Signal, QTimer
from PySide6.QtGui import QKeySequence, QShortcut from PySide6.QtGui import QKeySequence, QShortcut
from PySide6.QtWidgets import ( from PySide6.QtWidgets import (
QWidget, QVBoxLayout, QHBoxLayout, QToolButton, QLabel, QMessageBox, QSplitter, QWidget, QVBoxLayout, QHBoxLayout, QToolButton, QLabel, QMessageBox, QSplitter,
QFrame, QLineEdit, QFileDialog, QMenu, QFrame, QLineEdit, QFileDialog, QMenu, QInputDialog,
) )
from app.session_manager import Session, AUTH_KEY, AUTH_PASSWORD from app.session_manager import Session, AUTH_KEY, AUTH_PASSWORD
@@ -30,6 +30,12 @@ from app.dialogs.connect_prompt import ask_secret, confirm_host_key, confirm_hos
from app.services.ssh_target import detect_ssh_target from app.services.ssh_target import detect_ssh_target
from app.services.reconnect import compute_backoff_delay from app.services.reconnect import compute_backoff_delay
from app.snippets import Snippet, SnippetStore from app.snippets import Snippet, SnippetStore
from app.ai.service import KIND_EXPLAIN, KIND_DIAGNOSE, KIND_SOLUTION, KIND_GENERATE, AIRequestWorker
from app.ai.danger import check_dangerous
from app.ai.redaction import redact
from app.ui.ai_panel import AIPanel
from app.dialogs.ai_command_dialog import AICommandDialog
from app.dialogs.ai_prompts import ensure_ai_notice_acknowledged, confirm_sensitive_content, MASK, CANCEL
# sequences d'echappement ANSI, retirees des journaux pour qu'ils restent lisibles # sequences d'echappement ANSI, retirees des journaux pour qu'ils restent lisibles
ANSI_ESCAPE_RE = re.compile( ANSI_ESCAPE_RE = re.compile(
@@ -48,11 +54,15 @@ class BaseTerminalTab(QWidget):
def __init__( def __init__(
self, initial_status: str, snippet_store: SnippetStore | None = None, self, initial_status: str, snippet_store: SnippetStore | None = None,
broadcast=None, parent=None, broadcast=None, ai_provider_factory=None, credential_store=None, parent=None,
): ):
super().__init__(parent) super().__init__(parent)
self.snippet_store = snippet_store self.snippet_store = snippet_store
self.broadcast = broadcast self.broadcast = broadcast
self.ai_provider_factory = ai_provider_factory
self.credential_store = credential_store
self.ai_panel: AIPanel | None = None
self._ai_command_worker: AIRequestWorker | None = None
layout = QVBoxLayout(self) layout = QVBoxLayout(self)
layout.setContentsMargins(0, 0, 0, 0) layout.setContentsMargins(0, 0, 0, 0)
@@ -82,6 +92,13 @@ class BaseTerminalTab(QWidget):
self.snippets_btn.clicked.connect(self._show_snippets_menu) self.snippets_btn.clicked.connect(self._show_snippets_menu)
self.bar.addWidget(self.snippets_btn) self.bar.addWidget(self.snippets_btn)
self.ai_btn = QToolButton()
self.ai_btn.setText("🤖 Assistant IA")
self.ai_btn.setToolTip("Analyser la sortie selectionnee ou generer une commande (optionnel)")
self.ai_btn.setPopupMode(QToolButton.InstantPopup)
self.ai_btn.clicked.connect(self._show_ai_menu)
self.bar.addWidget(self.ai_btn)
self.log_btn = QToolButton() self.log_btn = QToolButton()
self.log_btn.setText("⏺ Journal") self.log_btn.setText("⏺ Journal")
self.log_btn.setToolTip("Enregistrer la session dans un fichier") self.log_btn.setToolTip("Enregistrer la session dans un fichier")
@@ -269,6 +286,141 @@ class BaseTerminalTab(QWidget):
elif self.connection: elif self.connection:
self.connection.send(data) self.connection.send(data)
# ---------- assistant IA (optionnel) ----------
def _show_ai_menu(self) -> None:
menu = self._build_ai_menu()
menu.exec(self.ai_btn.mapToGlobal(self.ai_btn.rect().bottomLeft()))
def _build_ai_menu(self) -> QMenu:
menu = QMenu(self)
has_selection = self.terminal.has_selection()
explain_action = menu.addAction("Expliquer cette sortie")
explain_action.setEnabled(has_selection)
explain_action.triggered.connect(self._ai_explain)
diagnose_action = menu.addAction("Diagnostiquer cette erreur")
diagnose_action.setEnabled(has_selection)
diagnose_action.triggered.connect(self._ai_diagnose)
solution_action = menu.addAction("Proposer une solution")
solution_action.setEnabled(has_selection)
solution_action.triggered.connect(self._ai_solution)
ask_action = menu.addAction("Poser une question...")
ask_action.setEnabled(has_selection)
ask_action.triggered.connect(self._ai_ask)
if not has_selection:
for action in (explain_action, diagnose_action, solution_action, ask_action):
action.setToolTip("Selectionnez d'abord du texte dans le terminal.")
menu.addSeparator()
generate_action = menu.addAction("Generer une commande...")
generate_action.triggered.connect(self._ai_generate_command)
menu.addSeparator()
panel_visible = self.ai_panel is not None and self.ai_panel.isVisible()
toggle_action = menu.addAction("Masquer le panneau IA" if panel_visible else "Afficher le panneau IA")
toggle_action.triggered.connect(self._toggle_ai_panel)
menu.addSeparator()
settings_action = menu.addAction("Parametres de l'assistant IA...")
settings_action.triggered.connect(self._open_ai_settings)
return menu
def _ensure_ai_panel(self) -> AIPanel:
if self.ai_panel is None:
self.ai_panel = AIPanel(self.ai_provider_factory)
self.splitter.addWidget(self.ai_panel)
# cree masque : sans cela, Qt affiche automatiquement un widget
# fraichement ajoute a un splitter deja visible, ce qui romprait
# l'etat "ferme par defaut" attendu du panneau.
self.ai_panel.hide()
return self.ai_panel
def _toggle_ai_panel(self) -> None:
panel = self._ensure_ai_panel()
panel.setVisible(not panel.isVisible())
def _open_ai_settings(self) -> None:
from app.dialogs.ai_settings_dialog import AISettingsDialog
AISettingsDialog(credential_store=self.credential_store, parent=self).exec()
def _ai_explain(self) -> None:
self._run_ai_selection_action(KIND_EXPLAIN, "Expliquer cette sortie")
def _ai_diagnose(self) -> None:
self._run_ai_selection_action(KIND_DIAGNOSE, "Diagnostiquer cette erreur")
def _ai_solution(self) -> None:
self._run_ai_selection_action(KIND_SOLUTION, "Proposer une solution")
def _run_ai_selection_action(self, kind: str, label: str) -> None:
selection = self.terminal.selected_text()
if not selection:
return
panel = self._ensure_ai_panel()
panel.run_request(kind, selection, label=label)
def _ai_ask(self) -> None:
selection = self.terminal.selected_text()
if not selection:
return
panel = self._ensure_ai_panel()
panel.prompt_question(context=selection)
def _ai_generate_command(self) -> None:
if self._ai_command_worker is not None and self._ai_command_worker.isRunning():
return
provider = self.ai_provider_factory() if self.ai_provider_factory else None
if provider is None:
QMessageBox.information(
self, "Assistant IA",
"L'assistant IA n'est pas configure (menu Assistant IA > Parametres de l'assistant IA...).",
)
return
request, ok = QInputDialog.getMultiLineText(
self, "Generer une commande", "Decrivez ce que vous voulez accomplir :",
)
if not ok or not request.strip():
return
request = request.strip()
if not ensure_ai_notice_acknowledged(self):
return
decision = confirm_sensitive_content(self, request)
if decision == CANCEL:
return
if decision == MASK:
request = redact(request)
self.ai_btn.setEnabled(False)
self.ai_btn.setText("🤖 Generation...")
self._ai_command_worker = AIRequestWorker(provider, KIND_GENERATE, request)
self._ai_command_worker.finished_ok.connect(self._on_ai_command_ok)
self._ai_command_worker.finished_error.connect(self._on_ai_command_error)
self._ai_command_worker.start()
def _reset_ai_generate_button(self) -> None:
self.ai_btn.setEnabled(True)
self.ai_btn.setText("🤖 Assistant IA")
def _on_ai_command_ok(self, response: str) -> None:
self._reset_ai_generate_button()
dangerous = [m.label for m in check_dangerous(response)]
dialog = AICommandDialog(response, dangerous_labels=dangerous, parent=self)
if dialog.exec() and dialog.insert_requested and dialog.command:
self.insert_snippet_text(dialog.command)
def _on_ai_command_error(self, message: str) -> None:
self._reset_ai_generate_button()
QMessageBox.critical(self, "Assistant IA", message)
def insert_snippet_text(self, text: str) -> None:
"""N'envoie jamais Entree : le texte est ecrit, a l'utilisateur de valider."""
if self.connection:
self.connection.send(text.encode("utf-8"))
# ---------- recherche dans l'historique ---------- # ---------- recherche dans l'historique ----------
def open_search(self) -> None: def open_search(self) -> None:
self.search_bar.show() self.search_bar.show()
@@ -290,6 +442,10 @@ class BaseTerminalTab(QWidget):
def close_connection(self) -> None: def close_connection(self) -> None:
self._closing = True # ne jamais reconnecter automatiquement une fermeture voulue self._closing = True # ne jamais reconnecter automatiquement une fermeture voulue
self.stop_logging() self.stop_logging()
if self.ai_panel is not None:
self.ai_panel.shutdown()
if self._ai_command_worker is not None and self._ai_command_worker.isRunning():
self._ai_command_worker.wait(2000)
if self.connection: if self.connection:
self.connection.close_connection() self.connection.close_connection()
self.connection.wait(2000) self.connection.wait(2000)
@@ -300,11 +456,12 @@ class SessionTabWidget(BaseTerminalTab):
def __init__( def __init__(
self, session: Session, secret: str, bastion_target=None, bastion_password: str = "", self, session: Session, secret: str, bastion_target=None, bastion_password: str = "",
snippet_store=None, broadcast=None, parent=None, snippet_store=None, broadcast=None, ai_provider_factory=None, credential_store=None, parent=None,
): ):
super().__init__( super().__init__(
f"○ Connexion a {session.username}@{session.host}...", f"○ Connexion a {session.username}@{session.host}...",
snippet_store=snippet_store, broadcast=broadcast, parent=parent, snippet_store=snippet_store, broadcast=broadcast,
ai_provider_factory=ai_provider_factory, credential_store=credential_store, parent=parent,
) )
self.session = session self.session = session
self._bastion_target = bastion_target self._bastion_target = bastion_target
@@ -594,10 +751,14 @@ class LocalTerminalTabWidget(BaseTerminalTab):
sftp_requested = Signal(object) # cible ssh detectee (ou None) sftp_requested = Signal(object) # cible ssh detectee (ou None)
def __init__(self, shell: str = SHELL_POWERSHELL, snippet_store=None, broadcast=None, parent=None): def __init__(
self, shell: str = SHELL_POWERSHELL, snippet_store=None, broadcast=None,
ai_provider_factory=None, credential_store=None, parent=None,
):
super().__init__( super().__init__(
f"○ Ouverture de {SHELL_LABELS.get(shell, shell)}...", f"○ Ouverture de {SHELL_LABELS.get(shell, shell)}...",
snippet_store=snippet_store, broadcast=broadcast, parent=parent, snippet_store=snippet_store, broadcast=broadcast,
ai_provider_factory=ai_provider_factory, credential_store=credential_store, parent=parent,
) )
self.shell = shell self.shell = shell

65
tests/test_ai_command_dialog.py Normal file
View File

@@ -0,0 +1,65 @@
"""AICommandDialog : extraction de la commande depuis la reponse de l'IA
(bloc de code s'il y en a un), et les 3 actions possibles (Copier / Inserer
dans le terminal / Annuler) - jamais d'execution automatique."""
import sys, pathlib
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
from PySide6.QtWidgets import QApplication
from app.dialogs.ai_command_dialog import AICommandDialog, extract_command
app = QApplication.instance() or QApplication(sys.argv)
print("=== 1. extract_command() : bloc de code present ===")
response = "Pour lister les gros fichiers :\n\n```bash\nfind / -size +100M\n```\n\nCela peut prendre du temps."
assert extract_command(response) == "find / -size +100M"
print(" OK")
print()
print("=== 2. extract_command() : bloc de code sans langage indique ===")
response = "```\ndf -h\n```"
assert extract_command(response) == "df -h"
print(" OK")
print()
print("=== 3. extract_command() : pas de bloc de code -> reponse entiere ===")
response = "Utilisez simplement : uptime"
assert extract_command(response) == "Utilisez simplement : uptime"
print(" OK")
print()
print("=== 4. Avertissement affiche seulement si la commande est dangereuse ===")
dialog_safe = AICommandDialog("```\ndf -h\n```")
assert dialog_safe.command == "df -h"
dialog_dangerous = AICommandDialog("```\nrm -rf /tmp/x\n```", dangerous_labels=["rm -rf (suppression recursive)"])
assert dialog_dangerous.command == "rm -rf /tmp/x"
print(" OK (pas de verification visuelle automatisee, juste la construction sans exception)")
print()
print("=== 5. 'Copier' place la commande (eventuellement corrigee) dans le presse-papiers ===")
dialog = AICommandDialog("```\necho bonjour\n```")
dialog.command_view.setPlainText("echo modifie")
dialog._copy()
assert QApplication.clipboard().text() == "echo modifie"
print(" OK")
print()
print("=== 6. 'Inserer dans le terminal' : accepte le dialogue, ne l'execute jamais ===")
dialog = AICommandDialog("```\nls -la\n```")
assert dialog.insert_requested is False
dialog._insert()
assert dialog.insert_requested is True
assert dialog.command == "ls -la"
assert dialog.result() == 1 # QDialog.Accepted
print(" OK")
print()
print("=== 7. 'Annuler' : rejette le dialogue, insert_requested reste False ===")
dialog = AICommandDialog("```\nls -la\n```")
dialog.cancel_btn.click()
assert dialog.insert_requested is False
assert dialog.result() == 0 # QDialog.Rejected
print(" OK")
print()
print("AI COMMAND DIALOG : TOUS LES TESTS OK")

57
tests/test_ai_danger.py Normal file
View File

@@ -0,0 +1,57 @@
"""Detection locale de commandes potentiellement destructives proposees par
l'assistant IA (jamais un blocage, seulement un avertissement)."""
import sys, pathlib
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
from app.ai.danger import is_dangerous, check_dangerous
print("=== 1. Commandes destructrices detectees ===")
dangerous_examples = [
"rm -rf /var/log/*",
"rm -fr /tmp/build",
"mkfs.ext4 /dev/sdb1",
"dd if=/dev/zero of=/dev/sda bs=1M",
"wipefs -a /dev/sdb",
"sudo shutdown -h now",
"sudo reboot",
"iptables -F",
"nft flush ruleset",
"DROP DATABASE production;",
"DELETE FROM users;",
"docker system prune -a --volumes",
"zpool destroy tank",
"lvremove /dev/vg0/lv0",
"vgremove vg0",
]
for cmd in dangerous_examples:
result = is_dangerous(cmd)
print(f" {'[X]' if result else '[ ]'} {cmd}")
assert result, f"aurait du etre detectee comme dangereuse : {cmd}"
print(" OK")
print()
print("=== 2. Commandes normales non signalees ===")
safe_examples = [
"df -h",
"systemctl status nginx",
"docker ps -a",
"ls -la /var/log",
"find /var -type f -printf '%s %p\\n' | sort -nr | head -20",
"SELECT * FROM users WHERE id = 1;",
"git rm --cached fichier.txt", # 'rm' present mais pas -rf : ne doit pas matcher
]
for cmd in safe_examples:
result = is_dangerous(cmd)
print(f" {'[X]' if result else '[ ]'} {cmd}")
assert not result, f"n'aurait pas du etre signalee : {cmd}"
print(" OK")
print()
print("=== 3. check_dangerous() renvoie une explication exploitable ===")
matches = check_dangerous("rm -rf /")
print(" correspondances:", [m.label for m in matches])
assert matches and "rm" in matches[0].label.lower()
print(" OK")
print()
print("AI DANGER : TOUS LES TESTS OK")

168
tests/test_ai_panel.py Normal file
View File

@@ -0,0 +1,168 @@
"""AIPanel : point d'entree unique des requetes IA d'un onglet - verifie que
la verification de securite (avis + secrets) est bien invoquee avant tout
envoi, que les reponses/erreurs mettent a jour la conversation, et qu'une
reponse perimee (generation obsolete) est ignoree."""
import sys, pathlib
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
from unittest.mock import patch
from PySide6.QtCore import QCoreApplication
from PySide6.QtWidgets import QApplication
from app.ai.provider import AIProvider, AIError
from app.ai.service import KIND_EXPLAIN, KIND_ASK
from app.dialogs.ai_prompts import MASK, SEND, CANCEL
from app.ui.ai_panel import AIPanel
app = QApplication.instance() or QApplication(sys.argv)
class FakeProvider(AIProvider):
def __init__(self, reply="reponse IA", fail_with=None):
self.reply = reply
self.fail_with = fail_with
self.chat_calls = []
def chat(self, messages, *, temperature=None):
self.chat_calls.append(messages)
if self.fail_with:
raise self.fail_with
return self.reply
def pump(ms=3000):
"""Laisse le worker en arriere-plan se terminer et livrer son signal."""
QCoreApplication.processEvents()
import time
deadline = time.time() + ms / 1000
while time.time() < deadline:
QCoreApplication.processEvents()
time.sleep(0.01)
NOTICE_OK = patch("app.ui.ai_panel.ensure_ai_notice_acknowledged", return_value=True)
SEND_ASIS = patch("app.ui.ai_panel.confirm_sensitive_content", return_value=SEND)
print("=== 1. Aucun fournisseur configure : message clair, pas de requete lancee ===")
panel = AIPanel(provider_factory=lambda: None)
panel.run_request(KIND_EXPLAIN, "erreur X", label="Expliquer cette sortie")
assert "non configure" in panel.conversation.toPlainText().lower()
assert panel._worker is None
print(" OK")
print()
print("=== 2. Avis de principe refuse : rien n'est envoye ===")
provider = FakeProvider()
panel = AIPanel(provider_factory=lambda: provider)
with patch("app.ui.ai_panel.ensure_ai_notice_acknowledged", return_value=False):
panel.run_request(KIND_EXPLAIN, "erreur X", label="Expliquer cette sortie")
assert provider.chat_calls == []
assert panel._worker is None
print(" OK")
print()
print("=== 3. Secret detecte + annulation : rien n'est envoye ===")
provider = FakeProvider()
panel = AIPanel(provider_factory=lambda: provider)
with NOTICE_OK, patch("app.ui.ai_panel.confirm_sensitive_content", return_value=CANCEL):
panel.run_request(KIND_EXPLAIN, "password=abc123", label="Expliquer cette sortie")
assert provider.chat_calls == []
print(" OK")
print()
print("=== 4. Secret detecte + masquage : le texte envoye est redige ===")
provider = FakeProvider(reply="analyse")
panel = AIPanel(provider_factory=lambda: provider)
with NOTICE_OK, patch("app.ui.ai_panel.confirm_sensitive_content", return_value=MASK):
panel.run_request(KIND_EXPLAIN, "password=abc123", label="Expliquer cette sortie")
pump()
sent_content = provider.chat_calls[0][-1]["content"]
print(" contenu envoye:", sent_content)
assert "abc123" not in sent_content
assert "[MASQUE]" in sent_content
assert panel._last_response == "analyse"
print(" OK")
print()
print("=== 5. Requete reussie : reponse affichee et memorisee (Copier reponse) ===")
provider = FakeProvider(reply="Tout va bien.")
panel = AIPanel(provider_factory=lambda: provider)
with NOTICE_OK, SEND_ASIS:
panel.run_request(KIND_EXPLAIN, "sortie normale", label="Expliquer cette sortie")
pump()
assert "Tout va bien." in panel.conversation.toPlainText()
panel._copy_last_response()
assert QApplication.clipboard().text() == "Tout va bien."
print(" OK")
print()
print("=== 6. Erreur du fournisseur : affichee dans la conversation, pas d'exception ===")
provider = FakeProvider(fail_with=AIError("cle API invalide"))
panel = AIPanel(provider_factory=lambda: provider)
with NOTICE_OK, SEND_ASIS:
panel.run_request(KIND_EXPLAIN, "sortie", label="Expliquer cette sortie")
pump()
assert "cle API invalide" in panel.conversation.toPlainText()
print(" OK")
print()
print("=== 7. Effacer la conversation : generation incrementee, contenu vide ===")
provider = FakeProvider(reply="reponse")
panel = AIPanel(provider_factory=lambda: provider)
with NOTICE_OK, SEND_ASIS:
panel.run_request(KIND_EXPLAIN, "sortie", label="Expliquer cette sortie")
pump()
gen_before = panel._generation
panel.clear_conversation()
assert panel._generation == gen_before + 1
assert panel.conversation.toPlainText() == ""
assert panel._last_response == ""
print(" OK")
print()
print("=== 8. Une reponse d'une generation perimee est ignoree (annulation cooperative) ===")
provider = FakeProvider(reply="reponse tardive")
panel = AIPanel(provider_factory=lambda: provider)
with NOTICE_OK, SEND_ASIS:
panel.run_request(KIND_EXPLAIN, "sortie", label="Expliquer cette sortie")
worker = panel._worker
panel.clear_conversation() # invalide la generation avant que le worker n'ait fini
worker.wait(3000)
QCoreApplication.processEvents()
assert "reponse tardive" not in panel.conversation.toPlainText()
print(" OK")
print()
print("=== 9. Question libre (panneau) : contexte joint, question affichee ===")
provider = FakeProvider(reply="42")
panel = AIPanel(provider_factory=lambda: provider)
panel.prompt_question(context="ligne de contexte")
assert panel.question_row.isVisible()
panel.question_edit.setText("pourquoi ?")
with NOTICE_OK, SEND_ASIS:
panel._send_question()
pump()
assert not panel.question_row.isVisible()
last_messages = provider.chat_calls[-1]
assert any("ligne de contexte" in m["content"] for m in last_messages)
assert last_messages[-1]["content"] == "pourquoi ?"
assert "42" in panel.conversation.toPlainText()
print(" OK")
print()
print("=== 10. Une requete deja en cours bloque la suivante (une a la fois) ===")
provider = FakeProvider(reply="lente")
panel = AIPanel(provider_factory=lambda: provider)
with NOTICE_OK, SEND_ASIS:
panel.run_request(KIND_EXPLAIN, "sortie 1", label="Expliquer cette sortie")
first_worker = panel._worker
panel.run_request(KIND_EXPLAIN, "sortie 2", label="Expliquer cette sortie")
assert panel._worker is first_worker, "la seconde requete n'aurait pas du demarrer un nouveau worker"
assert "deja en cours" in panel.busy_label.text().lower()
pump()
print(" OK")
print()
print("AI PANEL : TOUS LES TESTS OK")

96
tests/test_ai_prompts.py Normal file
View File

@@ -0,0 +1,96 @@
"""Confirmations de securite avant envoi a l'assistant IA : avis de principe
affiche une seule fois (persist dans app_settings, redirige vers un fichier
temporaire), et verification des secrets a chaque envoi. QMessageBox est
simule (exec()/clickedButton() patches) : aucune boite modale reelle ne doit
bloquer les tests."""
import sys, pathlib
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
import tempfile
from pathlib import Path
from unittest.mock import patch
from PySide6.QtWidgets import QApplication, QMessageBox
import app.app_settings as app_settings
from app.dialogs.ai_prompts import ensure_ai_notice_acknowledged, confirm_sensitive_content, MASK, SEND, CANCEL
app = QApplication.instance() or QApplication(sys.argv)
tmp = Path(tempfile.mkdtemp())
settings_redirect = patch.object(app_settings, "SETTINGS_FILE", tmp / "settings.json")
settings_redirect.start()
def click(button_text):
"""Simule un clic sur le bouton `button_text` d'une QMessageBox construite
avec addButton(), sans faire tourner de boucle d'evenements reelle."""
def fake_clicked(self):
for b in self.buttons():
if b.text() == button_text:
return b
return None
return fake_clicked
print("=== 1. Avis de principe : affiche la premiere fois, refus -> False ===")
with patch("app.dialogs.ai_prompts.QMessageBox.exec", return_value=QMessageBox.Cancel) as mocked_exec:
result = ensure_ai_notice_acknowledged(None)
assert mocked_exec.called
assert result is False
assert app_settings.get_setting("ai_notice_acknowledged", False) is False
print(" OK")
print()
print("=== 2. Avis de principe : acceptation -> True, persiste ===")
with patch("app.dialogs.ai_prompts.QMessageBox.exec", return_value=QMessageBox.Ok) as mocked_exec:
result = ensure_ai_notice_acknowledged(None)
assert mocked_exec.called
assert result is True
assert app_settings.get_setting("ai_notice_acknowledged", False) is True
print(" OK")
print()
print("=== 3. Avis de principe : deja accepte -> plus jamais affiche ===")
with patch("app.dialogs.ai_prompts.QMessageBox.exec") as mocked_exec:
result = ensure_ai_notice_acknowledged(None)
assert not mocked_exec.called, "la boite ne doit plus s'afficher une fois l'accord persiste"
assert result is True
print(" OK")
print()
print("=== 4. Aucun secret detecte : SEND direct, aucune boite affichee ===")
with patch("app.dialogs.ai_prompts.QMessageBox.exec") as mocked_exec:
result = confirm_sensitive_content(None, "systemctl status nginx")
assert not mocked_exec.called
assert result == SEND
print(" OK")
print()
print("=== 5. Secret detecte + 'Masquer automatiquement' -> MASK ===")
with patch("app.dialogs.ai_prompts.QMessageBox.exec", return_value=0), \
patch("app.dialogs.ai_prompts.QMessageBox.clickedButton", click("Masquer automatiquement")):
result = confirm_sensitive_content(None, "password=Sup3rSecret!")
assert result == MASK
print(" OK")
print()
print("=== 6. Secret detecte + 'Voir le contenu' -> SEND (envoi tel quel) ===")
with patch("app.dialogs.ai_prompts.QMessageBox.exec", return_value=0), \
patch("app.dialogs.ai_prompts.QMessageBox.clickedButton", click("Voir le contenu")):
result = confirm_sensitive_content(None, "password=Sup3rSecret!")
assert result == SEND
print(" OK")
print()
print("=== 7. Secret detecte + 'Annuler' -> CANCEL ===")
with patch("app.dialogs.ai_prompts.QMessageBox.exec", return_value=0), \
patch("app.dialogs.ai_prompts.QMessageBox.clickedButton", click("Annuler")):
result = confirm_sensitive_content(None, "password=Sup3rSecret!")
assert result == CANCEL
print(" OK")
settings_redirect.stop()
print()
print("AI PROMPTS : TOUS LES TESTS OK")

157
tests/test_ai_provider.py Normal file
View File

@@ -0,0 +1,157 @@
"""OpenAICompatibleProvider : requete HTTP construite correctement, et gestion
des erreurs (401, 429, timeout, reseau, reponse invalide) - urllib.request.urlopen
est entierement mocke, aucun appel reseau reel."""
import sys, pathlib
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
import io
import json
import urllib.error
from unittest.mock import patch, MagicMock
from app.ai.provider import AIProvider, AIError, OpenAICompatibleProvider
def fake_response(payload: dict):
m = MagicMock()
m.read.return_value = json.dumps(payload).encode("utf-8")
m.__enter__.return_value = m
m.__exit__.return_value = False
return m
def make_http_error(code: int, body: bytes = b"") -> urllib.error.HTTPError:
return urllib.error.HTTPError(url="http://x/chat/completions", code=code, msg="err",
hdrs=None, fp=io.BytesIO(body))
provider = OpenAICompatibleProvider(
base_url="https://api.example.com/v1", api_key="sk-test", model="gpt-test", timeout=5, temperature=0.2,
)
print("=== 1. Requete construite correctement (URL, methode, en-tetes, corps) ===")
with patch("app.ai.provider.urllib.request.urlopen") as mocked:
mocked.return_value = fake_response({"choices": [{"message": {"content": "bonjour"}}]})
result = provider.chat([{"role": "user", "content": "salut"}])
request = mocked.call_args.args[0]
print(" URL:", request.full_url)
print(" methode:", request.get_method())
print(" Authorization:", request.get_header("Authorization"))
body = json.loads(request.data.decode("utf-8"))
print(" corps:", body)
assert request.full_url == "https://api.example.com/v1/chat/completions"
assert request.get_method() == "POST"
assert request.get_header("Authorization") == "Bearer sk-test"
assert body["model"] == "gpt-test"
assert body["temperature"] == 0.2
assert body["messages"] == [{"role": "user", "content": "salut"}]
assert result == "bonjour"
print(" OK")
print()
print("=== 2. Erreur 401 -> message clair mentionnant la cle API ===")
with patch("app.ai.provider.urllib.request.urlopen", side_effect=make_http_error(401)):
try:
provider.chat([{"role": "user", "content": "x"}])
ok = False
except AIError as exc:
ok = True
print(" message:", exc)
assert "401" in str(exc) and "cle" in str(exc).lower()
assert ok
print(" OK")
print()
print("=== 3. Erreur 429 -> message clair mentionnant la limitation ===")
with patch("app.ai.provider.urllib.request.urlopen", side_effect=make_http_error(429)):
try:
provider.chat([{"role": "user", "content": "x"}])
ok = False
except AIError as exc:
ok = True
print(" message:", exc)
assert "429" in str(exc)
assert ok
print(" OK")
print()
print("=== 4. Timeout -> message clair, pas une exception brute ===")
timeout_error = urllib.error.URLError(reason=TimeoutError("timed out"))
with patch("app.ai.provider.urllib.request.urlopen", side_effect=timeout_error):
try:
provider.chat([{"role": "user", "content": "x"}])
ok = False
except AIError as exc:
ok = True
print(" message:", exc)
assert "delai" in str(exc).lower() or "temps" in str(exc).lower()
assert ok
print(" OK")
print()
print("=== 5. Erreur reseau (DNS, connexion refusee) -> message clair ===")
network_error = urllib.error.URLError(reason="nom ou service inconnu")
with patch("app.ai.provider.urllib.request.urlopen", side_effect=network_error):
try:
provider.chat([{"role": "user", "content": "x"}])
ok = False
except AIError as exc:
ok = True
print(" message:", exc)
assert ok
print(" OK")
print()
print("=== 6. Reponse invalide (JSON incoherent) -> message clair, pas de KeyError brut ===")
with patch("app.ai.provider.urllib.request.urlopen") as mocked:
mocked.return_value = fake_response({"unexpected": "shape"})
try:
provider.chat([{"role": "user", "content": "x"}])
ok = False
except AIError as exc:
ok = True
print(" message:", exc)
assert ok
print(" OK")
print()
print("=== 7. Sans cle API : pas d'en-tete Authorization ===")
no_key_provider = OpenAICompatibleProvider(base_url="http://localhost:8080/v1", api_key="", model="local-model")
with patch("app.ai.provider.urllib.request.urlopen") as mocked:
mocked.return_value = fake_response({"choices": [{"message": {"content": "ok"}}]})
no_key_provider.chat([{"role": "user", "content": "x"}])
request2 = mocked.call_args.args[0]
assert request2.get_header("Authorization") is None
print(" OK (utile pour un LLM local sans authentification)")
print()
print("=== 8. Les 4 usages metier construisent les bons messages (via chat()) ===")
captured = []
class SpyProvider(AIProvider):
def chat(self, messages, *, temperature=None):
captured.append(messages)
return "reponse"
spy = SpyProvider()
spy.explain_output("erreur X")
spy.diagnose("erreur Y")
spy.propose_solution("erreur Z")
spy.generate_command("trouver les gros fichiers")
spy.ask("pourquoi ?", context="ligne de contexte")
print(" nombre d'appels a chat():", len(captured))
assert len(captured) == 5
assert "erreur X" in captured[0][-1]["content"]
assert "Cause probable" in captured[1][0]["content"] # structure imposee au system prompt de diagnose
assert "erreur Z" in captured[2][-1]["content"]
assert captured[3][-1]["content"] == "trouver les gros fichiers"
assert "ligne de contexte" in captured[4][1]["content"]
assert captured[4][-1]["content"] == "pourquoi ?"
print(" OK")
print()
print("AI PROVIDER : TOUS LES TESTS OK")

60
tests/test_ai_redaction.py Normal file
View File

@@ -0,0 +1,60 @@
"""Detection/redaction de secrets avant envoi a l'assistant IA."""
import sys, pathlib
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
from app.ai.redaction import detect_secrets, redact
print("=== 1. Cle privee PEM detectee ===")
text = "-----BEGIN RSA PRIVATE KEY-----\nMIIEowIBAAKCAQEA...\n-----END RSA PRIVATE KEY-----"
found = detect_secrets(text)
print(" detecte:", [f.label for f in found])
assert any("privee" in f.label.lower() for f in found)
print(" OK")
print()
print("=== 2. Jeton Bearer detecte ===")
found = detect_secrets("Authorization: Bearer sk-abcdefghijklmnopqrstuvwxyz123456")
labels = [f.label for f in found]
print(" detecte:", labels)
assert any("bearer" in l.lower() for l in labels)
assert any("authorization" in l.lower() for l in labels)
print(" OK")
print()
print("=== 3. Cle AWS detectee ===")
found = detect_secrets("export AWS_ACCESS_KEY_ID=AKIAIOSFODNN7EXAMPLE")
assert any("aws" in f.label.lower() for f in found)
print(" OK")
print()
print("=== 4. password= / api_key= / token= detectes ===")
assert any("mot de passe" in f.label.lower() for f in detect_secrets("password=Sup3rSecret!"))
assert any("api" in f.label.lower() for f in detect_secrets("api_key=abcd1234"))
assert any("jeton" in f.label.lower() for f in detect_secrets("token: eyJhbGciOiJIUzI1NiJ9"))
print(" OK")
print()
print("=== 5. Chaine longue ressemblant a un secret ===")
found = detect_secrets("valeur=" + "a1b2c3d4e5f6g7h8i9j0" * 3)
assert any("longue" in f.label.lower() for f in found)
print(" OK")
print()
print("=== 6. Texte anodin : rien detecte ===")
assert detect_secrets("systemd[1]: nginx.service: Failed with result 'exit-code'") == []
assert detect_secrets("nginx: [emerg] bind() to 0.0.0.0:443 failed (98: Address already in use)") == []
print(" OK")
print()
print("=== 7. redact() masque sans lever, texte anodin inchange ===")
redacted = redact("password=abc123 et Bearer xyzxyzxyzxyzxyzxyz")
print(" resultat:", redacted)
assert "abc123" not in redacted
assert "xyzxyzxyzxyzxyzxyz" not in redacted
assert "[MASQUE]" in redacted
clean = "systemctl status nginx"
assert redact(clean) == clean
print(" OK")
print()
print("AI REDACTION : TOUS LES TESTS OK")

168
tests/test_ai_service.py Normal file
View File

@@ -0,0 +1,168 @@
"""AIRequestWorker : execution d'une requete IA dans un vrai QThread en
arriere-plan, jamais sur le thread Qt principal - avec dispatch correct selon
le type de requete et remontee des erreurs via signal plutot qu'exception."""
import sys, pathlib
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
import threading
from PySide6.QtCore import QCoreApplication
from PySide6.QtWidgets import QApplication
from app.ai.provider import AIProvider, AIError
from app.ai.service import (
AIRequestWorker, KIND_EXPLAIN, KIND_DIAGNOSE, KIND_SOLUTION, KIND_GENERATE, KIND_ASK, KIND_TEST,
)
app = QApplication.instance() or QApplication(sys.argv)
class RecordingProvider(AIProvider):
"""Enregistre la methode appelee et le thread d'execution, pour verifier
a la fois le dispatch correct et le fait que ca tourne hors du thread principal."""
def __init__(self, fail_with: Exception | None = None):
self.calls = []
self.threads = []
self.fail_with = fail_with
def chat(self, messages, *, temperature=None):
self.calls.append(("chat", messages))
self.threads.append(threading.current_thread())
if self.fail_with:
raise self.fail_with
return "reponse-chat"
def explain_output(self, output):
self.calls.append(("explain_output", output))
self.threads.append(threading.current_thread())
if self.fail_with:
raise self.fail_with
return "reponse-explain"
def diagnose(self, output):
self.calls.append(("diagnose", output))
self.threads.append(threading.current_thread())
if self.fail_with:
raise self.fail_with
return "reponse-diagnose"
def propose_solution(self, output):
self.calls.append(("propose_solution", output))
self.threads.append(threading.current_thread())
if self.fail_with:
raise self.fail_with
return "reponse-solution"
def generate_command(self, request):
self.calls.append(("generate_command", request))
self.threads.append(threading.current_thread())
if self.fail_with:
raise self.fail_with
return "reponse-generate"
def ask(self, question, context=""):
self.calls.append(("ask", (question, context)))
self.threads.append(threading.current_thread())
if self.fail_with:
raise self.fail_with
return "reponse-ask"
def run_worker(worker, timeout_ms=5000):
results = {}
def on_ok(text):
results["ok"] = text
def on_error(text):
results["error"] = text
worker.finished_ok.connect(on_ok)
worker.finished_error.connect(on_error)
worker.start()
worker.wait(timeout_ms)
# laisser la boucle d'evenements livrer les signaux en file d'attente (cross-thread)
QCoreApplication.processEvents()
return results
main_thread = threading.current_thread()
print("=== 1. Dispatch correct par 'kind' (explain/diagnose/solution/generate/ask/test) ===")
cases = [
(KIND_EXPLAIN, "explain_output", "sortie X", "reponse-explain"),
(KIND_DIAGNOSE, "diagnose", "erreur Y", "reponse-diagnose"),
(KIND_SOLUTION, "propose_solution", "probleme Z", "reponse-solution"),
(KIND_GENERATE, "generate_command", "trouver les gros fichiers", "reponse-generate"),
]
for kind, expected_method, payload, expected_reply in cases:
provider = RecordingProvider()
worker = AIRequestWorker(provider, kind, payload)
results = run_worker(worker)
print(f" kind={kind!r} -> methode appelee: {provider.calls[0][0]!r}, resultat: {results!r}")
assert provider.calls == [(expected_method, payload)]
assert results.get("ok") == expected_reply
assert "error" not in results
print(" OK")
print()
print("=== 2. KIND_ASK transmet aussi le contexte ===")
provider = RecordingProvider()
worker = AIRequestWorker(provider, KIND_ASK, "pourquoi ?", context="ligne de contexte")
results = run_worker(worker)
print(" appel:", provider.calls[0], "resultat:", results)
assert provider.calls == [("ask", ("pourquoi ?", "ligne de contexte"))]
assert results.get("ok") == "reponse-ask"
print(" OK")
print()
print("=== 3. KIND_TEST appelle chat() avec un message minimal ===")
provider = RecordingProvider()
worker = AIRequestWorker(provider, KIND_TEST, "")
results = run_worker(worker)
print(" appel:", provider.calls[0], "resultat:", results)
assert provider.calls[0][0] == "chat"
assert results.get("ok") == "reponse-chat"
print(" OK")
print()
print("=== 4. AIError est capturee et remontee via finished_error, jamais levee ===")
provider = RecordingProvider(fail_with=AIError("cle API invalide"))
worker = AIRequestWorker(provider, KIND_EXPLAIN, "sortie")
results = run_worker(worker)
print(" resultat:", results)
assert results.get("error") == "cle API invalide"
assert "ok" not in results
print(" OK")
print()
print("=== 5. Exception inattendue capturee aussi (jamais de crash silencieux du thread) ===")
provider = RecordingProvider(fail_with=RuntimeError("boum"))
worker = AIRequestWorker(provider, KIND_EXPLAIN, "sortie")
results = run_worker(worker)
print(" resultat:", results)
assert "error" in results and "boum" in results["error"]
print(" OK")
print()
print("=== 6. Type de requete inconnu -> erreur claire, pas de plantage ===")
provider = RecordingProvider()
worker = AIRequestWorker(provider, "kind-inexistant", "x")
results = run_worker(worker)
print(" resultat:", results)
assert "error" in results and "inconnu" in results["error"].lower()
print(" OK")
print()
print("=== 7. L'execution se fait bien dans un thread separe du thread principal ===")
provider = RecordingProvider()
worker = AIRequestWorker(provider, KIND_EXPLAIN, "sortie")
run_worker(worker)
print(" thread principal:", main_thread, "thread d'execution:", provider.threads[0])
assert provider.threads[0] is not main_thread
assert isinstance(worker, __import__("PySide6.QtCore", fromlist=["QThread"]).QThread)
print(" OK")
print()
print("AI SERVICE : TOUS LES TESTS OK")

99
tests/test_ai_settings.py Normal file
View File

@@ -0,0 +1,99 @@
"""Configuration de l'assistant IA : parametres non sensibles persistes via
app_settings (redirige vers un fichier temporaire pour ne jamais toucher les
vrais reglages), cle API exclusivement via CredentialStore, et construction du
fournisseur a partir des deux."""
import sys, pathlib
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
import tempfile
from pathlib import Path
from unittest.mock import patch
from app.credentials import CredentialStore
from app.ai.settings import (
AISettings, load_ai_settings, save_ai_settings, load_api_key, save_api_key,
build_provider_from_settings, AI_API_KEY_CREDENTIAL, DEFAULT_BASE_URL,
)
from app.ai.provider import OpenAICompatibleProvider
import app.app_settings as app_settings
class FakeCredentialStore(CredentialStore):
def __init__(self):
self._data = {}
def is_available(self):
return True
def set_secret(self, key, secret):
self._data[key] = secret
return True
def get_secret(self, key):
return self._data.get(key)
def delete_secret(self, key):
self._data.pop(key, None)
tmp = Path(tempfile.mkdtemp())
settings_redirect = patch.object(app_settings, "SETTINGS_FILE", tmp / "settings.json")
settings_redirect.start()
print("=== 1. Valeurs par defaut : desactive, pas de secret expose ===")
default = load_ai_settings()
print(" ", default)
assert default.enabled is False
assert default.base_url == DEFAULT_BASE_URL
print(" OK")
print()
print("=== 2. Sauvegarde/rechargement des parametres non sensibles ===")
settings = AISettings(enabled=True, provider_name="OpenRouter", base_url="https://openrouter.ai/api/v1",
model="anthropic/claude", timeout=45.0, temperature=0.7)
save_ai_settings(settings)
reloaded = load_ai_settings()
print(" ", reloaded)
assert reloaded == settings
print(" OK")
print()
print("=== 3. La cle API ne transite jamais par app_settings (verification du fichier disque) ===")
raw = (tmp / "settings.json").read_text(encoding="utf-8")
assert "sk-" not in raw
assert "api_key" not in raw.lower() or "ai_api" not in raw.lower() # aucune cle presente, seul le nom du champ pourrait
print(" contenu persiste:", raw)
print(" OK")
print()
print("=== 4. Cle API : lecture/ecriture via CredentialStore uniquement ===")
store = FakeCredentialStore()
assert load_api_key(store) == ""
save_api_key(store, "sk-test-secret")
assert load_api_key(store) == "sk-test-secret"
assert store.get_secret(AI_API_KEY_CREDENTIAL) == "sk-test-secret"
save_api_key(store, "") # vider la cle -> supprime l'entree
assert load_api_key(store) == ""
print(" OK")
print()
print("=== 5. build_provider_from_settings() : None si desactive ===")
save_ai_settings(AISettings(enabled=False))
assert build_provider_from_settings(store) is None
print(" OK")
print()
print("=== 6. build_provider_from_settings() : fournisseur pret si active ===")
save_ai_settings(AISettings(enabled=True, base_url="https://api.example.com/v1", model="gpt-x"))
save_api_key(store, "sk-live")
provider = build_provider_from_settings(store)
assert isinstance(provider, OpenAICompatibleProvider)
assert provider.base_url == "https://api.example.com/v1"
assert provider.model == "gpt-x"
assert provider.api_key == "sk-live"
print(" OK")
settings_redirect.stop()
print()
print("AI SETTINGS : TOUS LES TESTS OK")

160
tests/test_ai_tab_integration.py Normal file
View File

@@ -0,0 +1,160 @@
"""Integration de l'assistant IA dans un onglet terminal : menu (active/desactive
selon la selection), panneau lateral cree a la demande dans le splitter (meme
mecanisme que le panneau SFTP), et generation de commande (jamais executee
automatiquement, avec avertissement si dangereuse).
LocalTerminalConnection.start() est neutralisee comme dans test_snippets.py :
aucun sous-processus reel, send() depose seulement les octets dans une file."""
import sys, pathlib
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
from unittest.mock import patch
from PySide6.QtCore import QCoreApplication
from PySide6.QtWidgets import QApplication
from app.local_terminal import LocalTerminalConnection
from app.ai.provider import AIProvider
from app.dialogs.ai_prompts import SEND
from app.ui.ai_panel import AIPanel
from app.ui.tabs import LocalTerminalTabWidget
app = QApplication.instance() or QApplication(sys.argv)
NOTICE_OK = patch("app.ui.ai_panel.ensure_ai_notice_acknowledged", return_value=True)
SEND_ASIS = patch("app.ui.ai_panel.confirm_sensitive_content", return_value=SEND)
PROMPTS_OK_IN_TABS = patch("app.ui.tabs.ensure_ai_notice_acknowledged", return_value=True)
SEND_ASIS_IN_TABS = patch("app.ui.tabs.confirm_sensitive_content", return_value=SEND)
class FakeProvider(AIProvider):
def __init__(self, reply="reponse"):
self.reply = reply
def chat(self, messages, *, temperature=None):
return self.reply
def pump(ms=3000):
import time
deadline = time.time() + ms / 1000
while time.time() < deadline:
QCoreApplication.processEvents()
time.sleep(0.01)
def make_tab(ai_provider_factory=None):
return LocalTerminalTabWidget(ai_provider_factory=ai_provider_factory)
print("=== 1. Menu IA : actions de selection desactivees sans selection ===")
with patch.object(LocalTerminalConnection, "start", lambda self: None):
tab = make_tab()
assert not tab.terminal.has_selection()
menu = tab._build_ai_menu()
labels = [a.text() for a in menu.actions() if a.text()]
print(" actions:", labels)
assert "Expliquer cette sortie" in labels
assert "Generer une commande..." in labels
for text in ("Expliquer cette sortie", "Diagnostiquer cette erreur", "Proposer une solution", "Poser une question..."):
action = next(a for a in menu.actions() if a.text() == text)
assert not action.isEnabled(), f"{text} devrait etre desactive sans selection"
generate_action = next(a for a in menu.actions() if a.text() == "Generer une commande...")
assert generate_action.isEnabled(), "Generer une commande ne necessite pas de selection"
tab.close_connection()
print(" OK")
print()
print("=== 2. Menu IA : actions de selection activees des qu'il y a une selection ===")
with patch.object(LocalTerminalConnection, "start", lambda self: None):
tab = make_tab()
tab.terminal.feed(b"erreur : connection refused\r\n")
tab.terminal._sel_anchor, tab.terminal._sel_focus = (0, 0), (10, 0)
assert tab.terminal.has_selection()
menu = tab._build_ai_menu()
for text in ("Expliquer cette sortie", "Diagnostiquer cette erreur", "Proposer une solution", "Poser une question..."):
action = next(a for a in menu.actions() if a.text() == text)
assert action.isEnabled(), f"{text} devrait etre active avec une selection"
tab.close_connection()
print(" OK")
print()
print("=== 3. 'Expliquer cette sortie' : cree le panneau IA dans le splitter et lance la requete ===")
with patch.object(LocalTerminalConnection, "start", lambda self: None):
provider = FakeProvider(reply="Ceci indique une connexion refusee.")
tab = make_tab(ai_provider_factory=lambda: provider)
tab.terminal.feed(b"erreur : connection refused\r\n")
tab.terminal._sel_anchor, tab.terminal._sel_focus = (0, 0), (10, 0)
assert tab.ai_panel is None
with NOTICE_OK, SEND_ASIS:
tab._ai_explain()
pump()
assert isinstance(tab.ai_panel, AIPanel)
assert tab.splitter.indexOf(tab.ai_panel) >= 0
assert "connexion refusee" in tab.ai_panel.conversation.toPlainText().lower()
tab.close_connection()
print(" OK")
print()
print("=== 4. Basculer l'affichage du panneau IA (menu) ===")
with patch.object(LocalTerminalConnection, "start", lambda self: None):
tab = make_tab(ai_provider_factory=lambda: FakeProvider())
tab.show() # isVisible() se compose avec les ancetres : il en faut un reellement affiche
tab._toggle_ai_panel() # premiere fois : cree + affiche
assert tab.ai_panel is not None and tab.ai_panel.isVisible()
tab._toggle_ai_panel() # deuxieme fois : masque
assert not tab.ai_panel.isVisible()
tab.close_connection()
print(" OK")
print()
print("=== 5. Generer une commande : jamais auto-execute, insertion sans Entree ===")
with patch.object(LocalTerminalConnection, "start", lambda self: None):
provider = FakeProvider(reply="Pour lister les gros fichiers :\n\n```bash\nfind / -size +100M\n```")
tab = make_tab(ai_provider_factory=lambda: provider)
with PROMPTS_OK_IN_TABS, SEND_ASIS_IN_TABS, \
patch("app.ui.tabs.QInputDialog.getMultiLineText", return_value=("trouver les gros fichiers", True)), \
patch("app.ui.tabs.AICommandDialog.exec", lambda self: (setattr(self, "insert_requested", True), 1)[1]):
tab._ai_generate_command()
pump()
sent = tab.connection._out_queue.get_nowait()
print(" commande inseree:", sent)
assert sent == b"find / -size +100M"
assert not sent.endswith(b"\r") and not sent.endswith(b"\n")
assert tab.ai_btn.isEnabled() and tab.ai_btn.text() == "🤖 Assistant IA"
tab.close_connection()
print(" OK")
print()
print("=== 6. Generer une commande : annulee par l'utilisateur -> rien envoye ===")
with patch.object(LocalTerminalConnection, "start", lambda self: None):
provider = FakeProvider(reply="```bash\nrm -rf /\n```")
tab = make_tab(ai_provider_factory=lambda: provider)
with patch("app.ui.tabs.QInputDialog.getMultiLineText", return_value=("", False)):
tab._ai_generate_command()
assert tab.connection._out_queue.empty()
tab.close_connection()
print(" OK")
print()
print("=== 7. Sans fournisseur configure : message clair, pas de boite de saisie ===")
with patch.object(LocalTerminalConnection, "start", lambda self: None):
tab = make_tab(ai_provider_factory=lambda: None)
with patch("app.ui.tabs.QInputDialog.getMultiLineText") as mocked_input, \
patch("app.ui.tabs.QMessageBox.information") as mocked_info:
tab._ai_generate_command()
assert not mocked_input.called
assert mocked_info.called
tab.close_connection()
print(" OK")
print()
print("=== 8. Fermeture de l'onglet : le panneau IA est arrete proprement ===")
with patch.object(LocalTerminalConnection, "start", lambda self: None):
tab = make_tab(ai_provider_factory=lambda: FakeProvider())
tab._toggle_ai_panel()
tab.close_connection() # ne doit pas lever, ni bloquer indefiniment
print(" OK")
print()
print("AI TAB INTEGRATION : TOUS LES TESTS OK")