fix: Correction de la logique de changement de mot de passe et mise à jour des messages d'erreur

refactor: Amélioration de la clarté du code et gestion des erreurs. Séparation des responsabilités méthodologiques (routes et login/chag_pwd des utilisateurs). Création d'un service dédié pour le changement de mot de passe et le login utilisateur.

fix: Correction de la logique de changement de mot de passe et mise à jour des messages d'erreur

Author: remiv1

app_acfc/application.py

@@ -392,9 +392,9 @@ def login() -> Any:
         try:
             if not user_to_authenticate.authenticate():
                 return render_template(LOGIN['page'], title=LOGIN['title'], context=LOGIN['context'], message=INVALID)
-            elif user_to_authenticate.user_data.is_chg_mdp:
+            elif user_to_authenticate.is_chg_mdp:
                 return render_template(LOGIN['page'], title=LOGIN['title'], context='change_password',
-                                       message="Veuillez changer votre mot de passe.", username=user_to_authenticate.user_data.pseudo)
+                                       message="Veuillez changer votre mot de passe.", username=user_to_authenticate.user_pseudo)
             else:
                 return redirect(url_for('dashboard'))
         except Exception as e:
@@ -639,46 +639,24 @@ def chg_pwd() -> Any:
     """
     Changement de mot de passe utilisateur.
     """
+    def _get_key_message(param: Dict[str, bool]) -> str:
+        for key, value in param.items():
+            if value is False:
+                return key
+        return ''
+
+    # === Gestion de la méthode POST ===
     if request.method == 'POST':
         # Récupération des données du formulaire
-        username = request.form.get('username', '')
-        old_password = request.form.get('old_password', '')
-        new_password = request.form.get('new_password', '')
-        confirm_password = request.form.get('confirm_password', '')
-
-        # Validation des données
-        if not all([username, old_password, new_password, confirm_password]):
-            return render_template(CHG_PWD['page'], title=CHG_PWD['title'], context=CHG_PWD['context'],
-                                   message='Merci de remplir tous les champs.', username=username)
-
-        if new_password != confirm_password:
-            return render_template(CHG_PWD['page'], title=CHG_PWD['title'], context=CHG_PWD['context'],
-                                   message="Les mots de passe ne correspondent pas.", username=username)
-
-        if new_password == old_password:
-            return render_template(CHG_PWD['page'], title=CHG_PWD['title'], context=CHG_PWD['context'],
-                                   message="Le nouveau mot de passe ne peut pas être identique à l'ancien.", username=username)
-
-        # Vérification de l'ancien mot de passe
-        db_session = SessionBdD()
-        user = db_session.query(User).filter_by(pseudo=username).first()
-        if not user or not ph_acfc.verify_password(old_password, user.sha_mdp):
+        user_to_chg_pwd = AuthenticationService(request)
+        chg_pwd_is_ok = user_to_chg_pwd.chg_pwd()
+        if not chg_pwd_is_ok:
+            key_message = _get_key_message(user_to_chg_pwd.pwd_param)
             return render_template(CHG_PWD['page'], title=CHG_PWD['title'], context=CHG_PWD['context'],
-                                   message='Ancien mot de passe incorrect.', username=username)
-
-        # Hashage du nouveau mot de passe
-        user.sha_mdp = ph_acfc.hash_password(new_password)
-
-        # retrait de la nécessité de changer le mot de passe
-        user.is_chg_mdp = False
-        try:
-            db_session.commit()
-        except Exception as e:
-            db_session.rollback()
-            return render_template(CHG_PWD['page'], title=CHG_PWD['title'], context='500', message=str(e))
-
-        return redirect(url_for('login'))
-
+                                   message=user_to_chg_pwd.pwd_param_messages[key_message], username=user_to_chg_pwd.user)
+        else:
+            return redirect(url_for('login'))
+    # === Gestion de toutes les autres méthodes ===
     return render_template(ERROR400['page'], title=ERROR400['title'], context=ERROR400['context'], message=INVALID)
 
 # ====================================================================

app_acfc/services.py

@@ -19,9 +19,10 @@
 from argon2 import PasswordHasher
 from argon2.exceptions import VerifyMismatchError
 from flask import Flask, session
+from sqlalchemy.orm import Session as SessionBdDType
 from os import getenv
 from flask import Request
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Tuple
 from logs.logger import acfc_log, INFO
 
 LOG_LOGIN_FILE = 'login.log'
@@ -179,13 +180,27 @@ class AuthenticationService:
     - logout: Déconnecte l'utilisateur
 
     """
-    def __init__(self, request_object: Request):
+    def __init__(self, request_object: Request, chg_mdp: bool = False):
         self.user = request_object.form.get('username', '')
         self.pwd = request_object.form.get('password', '')
+        self.old_pwd = request_object.form.get('old_password', '')
+        self.new_pwd = request_object.form.get('new_password', '')
+        self.confirm_new_pwd = request_object.form.get('confirm_password', '')
         self.authenticated: bool = False
         self.user_dict: Dict[str, Any]
-        self.user_data: User
         self.existing_user: bool
+        self.is_chg_mdp: bool = False
+        self.user_pseudo: str = ""
+        self.pwd_param: Dict[str, bool] = {}
+        self.pwd_param_messages: Dict[str, str] = {
+            'all_field': 'Merci de remplir tous les champs.',
+            'pwd_match': "Les mots de passe ne correspondent pas.",
+            'pwd_changed': "Le nouveau mot de passe ne peut pas être identique à l'ancien.",
+            'authenticated_user': "Merci de renseigner correctement l'ancien mot de passe",
+            'db_unerror': "Erreur lors de la mise à jour en base de données.",
+            'user_found': "Utilisateur non trouvé."
+        }
+
 
     def _create_session(self, user: User):
         """
@@ -200,7 +215,7 @@ def _create_session(self, user: User):
         session['last_name'] = user.nom
         session['first_name'] = user.prenom
         session['email'] = user.email
-        session['tel'] = user.tel
+        session['tel'] = user.telephone
         session['habilitations'] = user.permission
 
     def _log_result(self, message: str, level: int = INFO):
@@ -216,6 +231,7 @@ def _log_result(self, message: str, level: int = INFO):
                             specific_logger=LOG_LOGIN_FILE,
                             zone_log=LOG_LOGIN_FILE,
                             db_log=True)
+
     def _bad_password(self, user: User):
         """
         Gère un mot de passe incorrect pour un utilisateur existant.
@@ -238,12 +254,23 @@ def _good_password(self, user: User, ph_acfc: PasswordService):
         self.existing_user = True
         self.authenticated = True
         self.user_dict = user.to_dict()
-        self.user_data = user
+        # Stocker les propriétés nécessaires avant de détacher l'objet de la session
+        self.is_chg_mdp = user.is_chg_mdp
+        acfc_log.log_to_file(level=INFO,
+                            message=f'Changement de mot de passe requis pour l\'utilisateur: {self.is_chg_mdp}',
+                            specific_logger=LOG_LOGIN_FILE,
+                            zone_log=LOG_LOGIN_FILE,
+                            db_log=True)
+        self.user_pseudo = user.pseudo
         user.nb_errors = 0
         user.is_locked = False
-        user.is_chg_mdp = False
-        if ph_acfc.needs_rehash(user.sha_mdp): user.sha_mdp = ph_acfc.hash_password(user.mot_de_passe)
+        if ph_acfc.needs_rehash(user.sha_mdp): user.sha_mdp = ph_acfc.hash_password(user.sha_mdp)
 
+    def _get_user(self) -> Tuple[User | None, SessionBdDType]:
+        from modeles import SessionBdD
+        session_db = SessionBdD()
+        user = session_db.query(User).filter_by(pseudo=self.user).first()
+        return user, session_db
 
     def authenticate(self) -> bool:
         """
@@ -255,16 +282,14 @@ def authenticate(self) -> bool:
         :Returns:
             None
         """
-        from modeles import SessionBdD
-        session_db = SessionBdD()
-        user = session_db.query(User).filter_by(pseudo=self.user).first()
+        user, session_db = self._get_user()
         # Instance du service de gestion des mots de passe (hachage Argon2)
         ph_acfc = PasswordService()
         try:
-            if user and ph_acfc.verify_password(self.pwd, user.mot_de_passe) and (not user.is_locked and user.is_active):
+            if user and ph_acfc.verify_password(self.pwd, user.sha_mdp) and (not user.is_locked and user.is_active):
                 self._good_password(user, ph_acfc)
                 self._create_session(user)
-                self._log_result(message=f'début de session pour l\'utilisateur: {self.user_data.pseudo if self.user_data else "inconnu"}')
+                self._log_result(message=f'début de session pour l\'utilisateur: {self.user_pseudo if self.user_pseudo else "inconnu"}')
                 statement = True
             elif user and user.is_locked:
                 self._log_result(message=f'Utilisateur verrouillé: {self.user}')
@@ -280,14 +305,65 @@ def authenticate(self) -> bool:
                 self._log_result(message=f'Utilisateur non trouvé: {self.user}')
                 statement = False
             session_db.commit()
-        except Exception:
+            # Détacher l'objet user de la session si il existe
+            if user is not None:
+                session_db.expunge(user)
+        except Exception as e:
             session_db.rollback()
-            self._log_result(message=f'Erreur lors de l\'authentification de l\'utilisateur: {self.user}', level=40)
+            self._log_result(message=f'Erreur lors de l\'authentification de l\'utilisateur: {self.user}.\nErreur: {e}', level=40)
             statement = False
         finally:
             # Fermeture de la session de base de données
             session_db.close()
         return statement
+    
+    def _validate_chg_pwd_form(self, user: User, pwd_hasher: PasswordService):
+        if not all([self.user, self.old_pwd, self.new_pwd, self.confirm_new_pwd]):
+            self.pwd_param['all_field'] = False
+            return False
+        elif self.new_pwd != self.confirm_new_pwd:
+            self.pwd_param['pwd_match'] = False
+            return False
+        elif self.new_pwd == self.old_pwd:
+            self.pwd_param['pwd_changed'] = False
+            return False
+        elif not pwd_hasher.verify_password(self.old_pwd, user.sha_mdp):
+            self.pwd_param['authenticated_user'] = False
+            return False
+        else:
+            self.pwd_param['all_field'] = True
+            self.pwd_param['pwd_match'] = True
+            self.pwd_param['pwd_changed'] = True
+            self.pwd_param['authenticated_user'] = True
+            return True
+
+    def chg_pwd(self) -> bool:
+        user, session_db = self._get_user()
+        ph_acfc = PasswordService()
+        if not self._validate_chg_pwd_form(user, ph_acfc):
+            return False
+        elif user:
+            try:
+                user.sha_mdp = ph_acfc.hash_password(self.new_pwd)
+                user.is_chg_mdp = False
+                user.nb_errors = 0
+                user.is_locked = False
+                session_db.commit()
+                self._log_result(message=f'Mot de passe changé pour l\'utilisateur: {self.user}.')
+                return True
+
+            except Exception as e:
+                session_db.rollback()
+                self._log_result(message=f'Erreur lors du changement de mot de passe de l\'utilisateur: {self.user}.\nErreur: {e}', level=40)
+                self.pwd_param['db_unerror'] = False
+                return False
+            
+            finally:
+                session_db.close()
+        else:
+            self._log_result(message=f'Utilisateur non trouvé: {self.user}.')
+            self.pwd_param['user_found'] = False
+            return False
 
     def is_authenticated(self) -> bool:
         """
@@ -296,4 +372,5 @@ def is_authenticated(self) -> bool:
         Returns:
             bool: True si l'utilisateur est authentifié, False sinon
         """
-        return self.authenticated
+        return self.authenticated
+    

mariadb/init_db.sql

@@ -22,7 +22,7 @@ CREATE TABLE IF NOT EXISTS `99_users` (
 INSERT INTO `99_users` (
     `prenom`, `nom`, `pseudo`, `sha_mdp`, `is_chg_mdp`, `email`, `telephone`, `is_active`, `permission`, `debut`, `nb_errors`, `is_locked`
 ) VALUES (
-    'Admin', 'istrateur', 'admin', '$argon2id$v=19$m=65536,t=4,p=3$5MRAo5AnOW3LV9gR/jRNFg$uWgIeoC6ZpyBowO/aNLTA2nndXfeGkEnsY+nsfCEzTc', TRUE, 'admin@example.com', '0000000000', TRUE, '1234567', CURRENT_DATE, 0, FALSE
+    'Admin', 'istrateur', 'admin', '$argon2id$v=19$m=65536,t=4,p=3$5MRAo5AnOW3LV9gR/jRNFg$uWgIeoC6ZpyBowO/aNLTA2nndXfeGkEnsY+nsfCEzTc', 1, 'admin@example.com', '0000000000', 1, '1234567', CURRENT_DATE, 0, 0
 )
 ON DUPLICATE KEY UPDATE
     `sha_mdp` = VALUES(`sha_mdp`),