adminmpmcz/patchcenter
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

Fix SSH audit: lire settings DB au lieu de hardcoder user/key/suffixes

Browse Source 
- SSH key et user lus depuis app_secrets (ssh_key_file, ssh_user)
- Ajout .mpcz.fr dans DNS_SUFFIXES
- Auto-detect Ed25519/RSA/ECDSA
- Fallback password depuis secrets
This commit is contained in:
Pierre & Lumière 2026-04-10 20:01:15 +02:00
parent c2f3d669eb
commit 16315ab3b5
1 changed files with 27 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
app/services/realtime_audit_service.py
View File

@ -11,10 +11,24 @@ try:
except ImportError: except ImportError:
PARAMIKO_OK = False PARAMIKO_OK = False
SSH_KEY = "/opt/patchcenter/keys/id_rsa_cybglobal.pem" SSH_KEY_DEFAULT = "/opt/patchcenter/keys/id_ed25519"
SSH_USER = "cybsecope" SSH_USER_DEFAULT = "root"
SSH_TIMEOUT = 12 SSH_TIMEOUT = 12
DNS_SUFFIXES = ["", ".sanef.groupe", ".sanef-rec.fr", ".sanef.fr"] DNS_SUFFIXES = ["", ".mpcz.fr", ".sanef.groupe", ".sanef-rec.fr", ".sanef.fr"]
def _get_ssh_settings():
"""Lit les settings SSH depuis app_secrets dans la DB."""
try:
from .secrets_service import get_secret
from ..database import SessionLocal
db = SessionLocal()
key_path = get_secret(db, "ssh_key_file") or SSH_KEY_DEFAULT
user = get_secret(db, "ssh_user") or SSH_USER_DEFAULT
db.close()
return key_path, user
except Exception:
return SSH_KEY_DEFAULT, SSH_USER_DEFAULT
# Commandes d'audit (simplifiees pour le temps reel) # Commandes d'audit (simplifiees pour le temps reel)
AUDIT_CMDS = { AUDIT_CMDS = {
@ -62,14 +76,16 @@ def _connect(target):
return None return None
import os import os
# 1. Essai clé SSH ssh_key, ssh_user = _get_ssh_settings()
if os.path.exists(SSH_KEY):
for loader in [paramiko.RSAKey.from_private_key_file, paramiko.Ed25519Key.from_private_key_file]: # 1. Essai clé SSH depuis settings
if os.path.exists(ssh_key):
for loader in [paramiko.Ed25519Key.from_private_key_file, paramiko.RSAKey.from_private_key_file, paramiko.ECDSAKey.from_private_key_file]:
try: try:
key = loader(SSH_KEY) key = loader(ssh_key)
client = paramiko.SSHClient() client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(target, port=22, username=SSH_USER, pkey=key, client.connect(target, port=22, username=ssh_user, pkey=key,
timeout=SSH_TIMEOUT, look_for_keys=False, allow_agent=False) timeout=SSH_TIMEOUT, look_for_keys=False, allow_agent=False)
return client return client
except Exception: except Exception:
@ -80,7 +96,7 @@ def _connect(target):
from .secrets_service import get_secret from .secrets_service import get_secret
from ..database import SessionLocal from ..database import SessionLocal
db = SessionLocal() db = SessionLocal()
pwd_user = get_secret(db, "ssh_pwd_default_user") or "root" pwd_user = get_secret(db, "ssh_pwd_default_user") or ssh_user
pwd_pass = get_secret(db, "ssh_pwd_default_pass") or "" pwd_pass = get_secret(db, "ssh_pwd_default_pass") or ""
db.close() db.close()
if pwd_pass: if pwd_pass:
@ -138,7 +154,8 @@ def audit_single_server(hostname):
return result return result
result["status"] = "OK" result["status"] = "OK"
result["connection_method"] = f"ssh_key ({SSH_USER}@{target})" ssh_key, ssh_user = _get_ssh_settings()
result["connection_method"] = f"ssh_key ({ssh_user}@{target})"
for key, cmd in AUDIT_CMDS.items(): for key, cmd in AUDIT_CMDS.items():
result[key] = _run(client, cmd) result[key] = _run(client, cmd)