From 7965925fad27625fd4dda16ba38b629c0854e60b Mon Sep 17 00:00:00 2001 From: root Date: Fri, 14 Aug 2026 17:33:05 +0000 Subject: [PATCH] Configure Yeastar credentials from the admin UI --- app_new/yeastar/models.py | 34 +++++------ app_new/yeastar/routes.py | 43 ++++++++++---- app_new/yeastar/templates/yeastar/config.html | 8 ++- tests/integration/test_yeastar_config.py | 35 ++++++++++++ watchdog_dnd.py | 57 ++++++++++++------- 5 files changed, 125 insertions(+), 52 deletions(-) create mode 100644 tests/integration/test_yeastar_config.py diff --git a/app_new/yeastar/models.py b/app_new/yeastar/models.py index 155d249..8531998 100644 --- a/app_new/yeastar/models.py +++ b/app_new/yeastar/models.py @@ -5,7 +5,6 @@ Surveillance téléphone P520 from datetime import datetime, timezone from app_new.extensions import db from cryptography.fernet import Fernet -import base64 import os @@ -24,41 +23,36 @@ class YeastarConfig(db.Model): # Clé de chiffrement (stockée en variable d'environnement ou générée) @staticmethod - def _get_encryption_key(): - """Récupère ou génère la clé de chiffrement.""" - key = os.environ.get('YEASTAR_ENCRYPTION_KEY') + def _get_fernet(): + """Construit Fernet avec une clé persistante fournie par l'environnement.""" + key = os.environ.get('YEASTAR_ENCRYPTION_KEY') or os.environ.get( + 'OUTLOOK_ENCRYPTION_KEY' + ) if not key: - # Clé par défaut (en production, utiliser une variable d'environnement) - key = 'Y2xhY2tfc2VjcmV0X2tleV9mb3JfeWVhc3Rhcl8yMDI0' - return base64.urlsafe_b64encode(key.encode()[:32].ljust(32, b'0')) + raise RuntimeError( + 'YEASTAR_ENCRYPTION_KEY ou OUTLOOK_ENCRYPTION_KEY doit etre configuree' + ) + return Fernet(key.encode() if isinstance(key, str) else key) def set_password(self, password): """Chiffre et stocke le mot de passe.""" - key = self._get_encryption_key() - f = Fernet(key) - self.password_encrypted = f.encrypt(password.encode()) + self.password_encrypted = self._get_fernet().encrypt(password.encode()) def get_password(self): """Déchiffre et retourne le mot de passe.""" if not self.password_encrypted: return None - key = self._get_encryption_key() - f = Fernet(key) - return f.decrypt(self.password_encrypted).decode() + return self._get_fernet().decrypt(self.password_encrypted).decode() def set_username(self, username): """Chiffre et stocke le nom d'utilisateur.""" - key = self._get_encryption_key() - f = Fernet(key) - self.username_encrypted = f.encrypt(username.encode()) + self.username_encrypted = self._get_fernet().encrypt(username.encode()) def get_username(self): """Déchiffre et retourne le nom d'utilisateur.""" if not self.username_encrypted: return None - key = self._get_encryption_key() - f = Fernet(key) - return f.decrypt(self.username_encrypted).decode() + return self._get_fernet().decrypt(self.username_encrypted).decode() @staticmethod def get_config(): @@ -92,4 +86,4 @@ class YeastarDNDConfig(db.Model): return getattr(self, key, default) def __repr__(self): - return f"" \ No newline at end of file + return f"" diff --git a/app_new/yeastar/routes.py b/app_new/yeastar/routes.py index e39a559..d1acc5d 100644 --- a/app_new/yeastar/routes.py +++ b/app_new/yeastar/routes.py @@ -8,6 +8,7 @@ from app_new.extensions import db from app_new.core.system_security import system_admin_required from app_new.yeastar.models import YeastarConfig, YeastarDNDConfig from app_new.lib_ext.api_client import YeastarClient +from cryptography.fernet import InvalidToken import subprocess from datetime import datetime, timezone @@ -21,8 +22,11 @@ def get_client(): if not config: return None, "Configuration Yeastar non trouvée. Veuillez configurer les identifiants." - username = config.get_username() - password = config.get_password() + try: + username = config.get_username() + password = config.get_password() + except (InvalidToken, ValueError, RuntimeError): + return None, "Les identifiants Yeastar doivent etre saisis de nouveau." if not all([config.base_url, username, password]): return None, "Identifiants incomplets dans la configuration." @@ -104,7 +108,7 @@ def index(): @yeastar_bp.route('/config', methods=['GET', 'POST']) -@login_required +@system_admin_required def config_page(): """Page de configuration des identifiants Yeastar.""" config = YeastarConfig.get_config() @@ -114,6 +118,10 @@ def config_page(): username = request.form.get('username', '').strip() password = request.form.get('password', '').strip() + if not base_url.startswith(('https://', 'http://')): + flash("L'URL Yeastar doit commencer par https:// ou http://.", "danger") + return redirect(url_for('yeastar.config_page')) + if not all([base_url, username]): flash("L'URL et le nom d'utilisateur sont obligatoires.", "danger") return redirect(url_for('yeastar.config_page')) @@ -129,10 +137,17 @@ def config_page(): config.set_username(username) if password: config.set_password(password) - elif not config.password_encrypted: - # Mot de passe obligatoire à la première création - flash("Le mot de passe est obligatoire pour la première configuration.", "danger") - return redirect(url_for('yeastar.config_page')) + else: + try: + existing_password = config.get_password() + except (InvalidToken, ValueError, RuntimeError): + existing_password = None + if not existing_password: + flash( + "Le mot de passe est obligatoire pour cette configuration.", + "danger", + ) + return redirect(url_for('yeastar.config_page')) config.updated_at = datetime.now(timezone.utc) db.session.commit() @@ -142,19 +157,27 @@ def config_page(): # Préparer les données pour l'affichage (ne pas afficher le mot de passe) display_config = None + password_is_valid = False if config: + try: + username = config.get_username() + config.get_password() + password_is_valid = True + except (InvalidToken, ValueError, RuntimeError): + username = '' display_config = { 'base_url': config.base_url, - 'username': config.get_username(), + 'username': username, 'is_active': config.is_active, - 'updated_at': config.updated_at + 'updated_at': config.updated_at, + 'password_is_valid': password_is_valid, } return render_template('yeastar/config.html', config=display_config) @yeastar_bp.route('/config/test', methods=['POST']) -@login_required +@system_admin_required def test_connection(): """Teste la connexion Yeastar avec les identifiants actuels.""" client, error = get_client() diff --git a/app_new/yeastar/templates/yeastar/config.html b/app_new/yeastar/templates/yeastar/config.html index 3a52025..585f8d1 100644 --- a/app_new/yeastar/templates/yeastar/config.html +++ b/app_new/yeastar/templates/yeastar/config.html @@ -46,10 +46,12 @@ + {% if not config or not config.password_is_valid %}required{% endif %}>
- {% if config %} + {% if config and config.password_is_valid %} Le mot de passe stocké est chiffré. Laissez vide pour le conserver. + {% elif config %} + La clé de chiffrement a changé : saisissez à nouveau le mot de passe. {% else %} Le mot de passe sera chiffré avant d'être stocké. {% endif %} @@ -161,4 +163,4 @@ document.getElementById('test-btn').addEventListener('click', function() { }); }); -{% endblock %} \ No newline at end of file +{% endblock %} diff --git a/tests/integration/test_yeastar_config.py b/tests/integration/test_yeastar_config.py new file mode 100644 index 0000000..75dfb11 --- /dev/null +++ b/tests/integration/test_yeastar_config.py @@ -0,0 +1,35 @@ +from app_new.yeastar.models import YeastarConfig + + +def test_admin_can_save_encrypted_yeastar_credentials(app, authenticated_client): + response = authenticated_client.post( + '/yeastar/config', + data={ + 'base_url': 'https://pbx.example.test', + 'username': 'maintenance', + 'password': 'a-strong-test-password', + }, + ) + assert response.status_code == 302 + + with app.app_context(): + config = YeastarConfig.get_config() + assert config.base_url == 'https://pbx.example.test' + assert config.get_username() == 'maintenance' + assert config.get_password() == 'a-strong-test-password' + assert b'maintenance' not in config.username_encrypted + assert b'a-strong-test-password' not in config.password_encrypted + + +def test_yeastar_url_requires_http_scheme(app, authenticated_client): + response = authenticated_client.post( + '/yeastar/config', + data={ + 'base_url': 'pbx.example.test', + 'username': 'maintenance', + 'password': 'a-strong-test-password', + }, + follow_redirects=True, + ) + assert response.status_code == 200 + assert b'doit commencer par https:// ou http://' in response.data diff --git a/watchdog_dnd.py b/watchdog_dnd.py index b40dabe..8b425e0 100644 --- a/watchdog_dnd.py +++ b/watchdog_dnd.py @@ -1,7 +1,6 @@ """Watchdog DND P520 - Surveillance continue - version autonome.""" import base64 import hashlib -import os import sys import time import uuid @@ -11,6 +10,8 @@ warnings.filterwarnings('ignore', message='Unverified HTTPS request') import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) +_flask_app = None + class YeastarClient: """Client minimal pour l'API Yeastar P-Series (P520)""" @@ -97,24 +98,42 @@ class YeastarClient: return False def load_config(): - """Charge la configuration depuis l'environnement, sans secret dans le code.""" - base_url = os.environ.get('YEASTAR_BASE_URL', '').strip() - username = os.environ.get('YEASTAR_USERNAME', '').strip() - password = os.environ.get('YEASTAR_PASSWORD', '') - explicitly_enabled = os.environ.get('YEASTAR_DND_ENABLED', '0') == '1' - return { - 'enabled': explicitly_enabled and bool(base_url and username and password), - 'refresh_seconds': int(os.environ.get('YEASTAR_DND_REFRESH_SECONDS', '2')), - 'reset_delay_seconds': int(os.environ.get('YEASTAR_DND_RESET_DELAY_SECONDS', '3')), - 'excluded_extensions': [ - value.strip() - for value in os.environ.get('YEASTAR_DND_EXCLUDED_EXTENSIONS', '').split(',') - if value.strip() - ], - 'base_url': base_url, - 'username': username, - 'password': password, - } + """Charge la configuration chiffree geree depuis l'interface GMAO.""" + global _flask_app + try: + if _flask_app is None: + from app_new import create_app + _flask_app = create_app() + from app_new.yeastar.models import YeastarConfig, YeastarDNDConfig + + with _flask_app.app_context(): + connection = YeastarConfig.get_config() + dnd = YeastarDNDConfig.query.first() + if not connection or not dnd: + raise RuntimeError('configuration absente') + username = connection.get_username() + password = connection.get_password() + return { + 'enabled': bool( + dnd.is_enabled and connection.base_url and username and password + ), + 'refresh_seconds': max(dnd.refresh_seconds or 2, 1), + 'reset_delay_seconds': max(dnd.reset_delay_seconds or 3, 0), + 'excluded_extensions': dnd.excluded_list, + 'base_url': connection.base_url, + 'username': username, + 'password': password, + } + except Exception: + return { + 'enabled': False, + 'refresh_seconds': 300, + 'reset_delay_seconds': 3, + 'excluded_extensions': [], + 'base_url': '', + 'username': '', + 'password': '', + } def log_dnd(message, level='info'): from datetime import datetime