Configure Yeastar credentials from the admin UI
Some checks are pending
CI - Tests et Syntax / lint-and-test (push) Waiting to run

This commit is contained in:
root 2026-08-14 17:33:05 +00:00
parent ef4e4ed3b4
commit 7965925fad
5 changed files with 125 additions and 52 deletions

View file

@ -5,7 +5,6 @@ Surveillance téléphone P520
from datetime import datetime, timezone from datetime import datetime, timezone
from app_new.extensions import db from app_new.extensions import db
from cryptography.fernet import Fernet from cryptography.fernet import Fernet
import base64
import os import os
@ -24,41 +23,36 @@ class YeastarConfig(db.Model):
# Clé de chiffrement (stockée en variable d'environnement ou générée) # Clé de chiffrement (stockée en variable d'environnement ou générée)
@staticmethod @staticmethod
def _get_encryption_key(): def _get_fernet():
"""Récupère ou génère la clé de chiffrement.""" """Construit Fernet avec une clé persistante fournie par l'environnement."""
key = os.environ.get('YEASTAR_ENCRYPTION_KEY') key = os.environ.get('YEASTAR_ENCRYPTION_KEY') or os.environ.get(
'OUTLOOK_ENCRYPTION_KEY'
)
if not key: if not key:
# Clé par défaut (en production, utiliser une variable d'environnement) raise RuntimeError(
key = 'Y2xhY2tfc2VjcmV0X2tleV9mb3JfeWVhc3Rhcl8yMDI0' 'YEASTAR_ENCRYPTION_KEY ou OUTLOOK_ENCRYPTION_KEY doit etre configuree'
return base64.urlsafe_b64encode(key.encode()[:32].ljust(32, b'0')) )
return Fernet(key.encode() if isinstance(key, str) else key)
def set_password(self, password): def set_password(self, password):
"""Chiffre et stocke le mot de passe.""" """Chiffre et stocke le mot de passe."""
key = self._get_encryption_key() self.password_encrypted = self._get_fernet().encrypt(password.encode())
f = Fernet(key)
self.password_encrypted = f.encrypt(password.encode())
def get_password(self): def get_password(self):
"""Déchiffre et retourne le mot de passe.""" """Déchiffre et retourne le mot de passe."""
if not self.password_encrypted: if not self.password_encrypted:
return None return None
key = self._get_encryption_key() return self._get_fernet().decrypt(self.password_encrypted).decode()
f = Fernet(key)
return f.decrypt(self.password_encrypted).decode()
def set_username(self, username): def set_username(self, username):
"""Chiffre et stocke le nom d'utilisateur.""" """Chiffre et stocke le nom d'utilisateur."""
key = self._get_encryption_key() self.username_encrypted = self._get_fernet().encrypt(username.encode())
f = Fernet(key)
self.username_encrypted = f.encrypt(username.encode())
def get_username(self): def get_username(self):
"""Déchiffre et retourne le nom d'utilisateur.""" """Déchiffre et retourne le nom d'utilisateur."""
if not self.username_encrypted: if not self.username_encrypted:
return None return None
key = self._get_encryption_key() return self._get_fernet().decrypt(self.username_encrypted).decode()
f = Fernet(key)
return f.decrypt(self.username_encrypted).decode()
@staticmethod @staticmethod
def get_config(): def get_config():

View file

@ -8,6 +8,7 @@ from app_new.extensions import db
from app_new.core.system_security import system_admin_required from app_new.core.system_security import system_admin_required
from app_new.yeastar.models import YeastarConfig, YeastarDNDConfig from app_new.yeastar.models import YeastarConfig, YeastarDNDConfig
from app_new.lib_ext.api_client import YeastarClient from app_new.lib_ext.api_client import YeastarClient
from cryptography.fernet import InvalidToken
import subprocess import subprocess
from datetime import datetime, timezone from datetime import datetime, timezone
@ -21,8 +22,11 @@ def get_client():
if not config: if not config:
return None, "Configuration Yeastar non trouvée. Veuillez configurer les identifiants." return None, "Configuration Yeastar non trouvée. Veuillez configurer les identifiants."
try:
username = config.get_username() username = config.get_username()
password = config.get_password() 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]): if not all([config.base_url, username, password]):
return None, "Identifiants incomplets dans la configuration." return None, "Identifiants incomplets dans la configuration."
@ -104,7 +108,7 @@ def index():
@yeastar_bp.route('/config', methods=['GET', 'POST']) @yeastar_bp.route('/config', methods=['GET', 'POST'])
@login_required @system_admin_required
def config_page(): def config_page():
"""Page de configuration des identifiants Yeastar.""" """Page de configuration des identifiants Yeastar."""
config = YeastarConfig.get_config() config = YeastarConfig.get_config()
@ -114,6 +118,10 @@ def config_page():
username = request.form.get('username', '').strip() username = request.form.get('username', '').strip()
password = request.form.get('password', '').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]): if not all([base_url, username]):
flash("L'URL et le nom d'utilisateur sont obligatoires.", "danger") flash("L'URL et le nom d'utilisateur sont obligatoires.", "danger")
return redirect(url_for('yeastar.config_page')) return redirect(url_for('yeastar.config_page'))
@ -129,9 +137,16 @@ def config_page():
config.set_username(username) config.set_username(username)
if password: if password:
config.set_password(password) config.set_password(password)
elif not config.password_encrypted: else:
# Mot de passe obligatoire à la première création try:
flash("Le mot de passe est obligatoire pour la première configuration.", "danger") 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')) return redirect(url_for('yeastar.config_page'))
config.updated_at = datetime.now(timezone.utc) config.updated_at = datetime.now(timezone.utc)
@ -142,19 +157,27 @@ def config_page():
# Préparer les données pour l'affichage (ne pas afficher le mot de passe) # Préparer les données pour l'affichage (ne pas afficher le mot de passe)
display_config = None display_config = None
password_is_valid = False
if config: if config:
try:
username = config.get_username()
config.get_password()
password_is_valid = True
except (InvalidToken, ValueError, RuntimeError):
username = ''
display_config = { display_config = {
'base_url': config.base_url, 'base_url': config.base_url,
'username': config.get_username(), 'username': username,
'is_active': config.is_active, '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) return render_template('yeastar/config.html', config=display_config)
@yeastar_bp.route('/config/test', methods=['POST']) @yeastar_bp.route('/config/test', methods=['POST'])
@login_required @system_admin_required
def test_connection(): def test_connection():
"""Teste la connexion Yeastar avec les identifiants actuels.""" """Teste la connexion Yeastar avec les identifiants actuels."""
client, error = get_client() client, error = get_client()

View file

@ -46,10 +46,12 @@
<label for="password" class="form-label">Mot de passe</label> <label for="password" class="form-label">Mot de passe</label>
<input type="password" class="form-control" id="password" name="password" <input type="password" class="form-control" id="password" name="password"
placeholder="{% if config %}Laisser vide pour conserver le mot de passe actuel{% else %}Mot de passe{% endif %}" placeholder="{% if config %}Laisser vide pour conserver le mot de passe actuel{% else %}Mot de passe{% endif %}"
{% if not config %}required{% endif %}> {% if not config or not config.password_is_valid %}required{% endif %}>
<div class="form-text text-muted"> <div class="form-text text-muted">
{% if config %} {% if config and config.password_is_valid %}
Le mot de passe stocké est chiffré. Laissez vide pour le conserver. 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 %} {% else %}
Le mot de passe sera chiffré avant d'être stocké. Le mot de passe sera chiffré avant d'être stocké.
{% endif %} {% endif %}

View file

@ -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

View file

@ -1,7 +1,6 @@
"""Watchdog DND P520 - Surveillance continue - version autonome.""" """Watchdog DND P520 - Surveillance continue - version autonome."""
import base64 import base64
import hashlib import hashlib
import os
import sys import sys
import time import time
import uuid import uuid
@ -11,6 +10,8 @@ warnings.filterwarnings('ignore', message='Unverified HTTPS request')
import urllib3 import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
_flask_app = None
class YeastarClient: class YeastarClient:
"""Client minimal pour l'API Yeastar P-Series (P520)""" """Client minimal pour l'API Yeastar P-Series (P520)"""
@ -97,24 +98,42 @@ class YeastarClient:
return False return False
def load_config(): def load_config():
"""Charge la configuration depuis l'environnement, sans secret dans le code.""" """Charge la configuration chiffree geree depuis l'interface GMAO."""
base_url = os.environ.get('YEASTAR_BASE_URL', '').strip() global _flask_app
username = os.environ.get('YEASTAR_USERNAME', '').strip() try:
password = os.environ.get('YEASTAR_PASSWORD', '') if _flask_app is None:
explicitly_enabled = os.environ.get('YEASTAR_DND_ENABLED', '0') == '1' 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 { return {
'enabled': explicitly_enabled and bool(base_url and username and password), 'enabled': bool(
'refresh_seconds': int(os.environ.get('YEASTAR_DND_REFRESH_SECONDS', '2')), dnd.is_enabled and connection.base_url and username and password
'reset_delay_seconds': int(os.environ.get('YEASTAR_DND_RESET_DELAY_SECONDS', '3')), ),
'excluded_extensions': [ 'refresh_seconds': max(dnd.refresh_seconds or 2, 1),
value.strip() 'reset_delay_seconds': max(dnd.reset_delay_seconds or 3, 0),
for value in os.environ.get('YEASTAR_DND_EXCLUDED_EXTENSIONS', '').split(',') 'excluded_extensions': dnd.excluded_list,
if value.strip() 'base_url': connection.base_url,
],
'base_url': base_url,
'username': username, 'username': username,
'password': password, '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'): def log_dnd(message, level='info'):
from datetime import datetime from datetime import datetime