blackify

app/auth/email.py

@@ -1,8 +1,80 @@
 # -*- coding: UTF-8 -*
+import re
+
 from flask import render_template, current_app
-from flask_babel import _
+
+from app import ScoValueError
 from app.email import send_email
 
+#  Fonctions de manipulation des adresses mail
+#  Objectif: traiter puis élminier les adresses mail doublonnées
+# pour pouvoir , à terme, les utilisaer commen identifiant pour inclusion d'un LDAP comme outil d'identification
+
+# principe: les adresses mails doublonnées superflues(note) sont transformées/numérotées sous une forme non autorisée
+# * les comptes ainsi modifiés seront désactivés et déclencheront une demande d'action auprès de l'utilisateur et/ou
+#   administrateur
+# * une notice est ajouté à la page des utlisateurs pour indiquer à l administrateur l'existence (et
+#   la liste) de doublons. Un bouton qui traite ces doublons est ajouté à cette liste.
+# note: une des adresses doublonnée sera conservée sous sa forme initiale (heuristique: on prend l'adresse du compte
+#    actif utilisé le plus récement)
+
+# transformation proposée: * un marqueur ( '.' initial normalement interdit dans les adresses mail est ajouté un
+# numéro d'ordre (+xxx) est ajouté à la partie locale de l'adresse
+# ainsi les trois adresses identiques u@domaine.org seront renommés en:
+#  u@domaine.org, .u+2@domaine.org et .u@domaine+3.org (la numérotation commence à 2)
+# cette transformation peut être redéfinie en remplaçant les fonctions check_email, transform_email et initial_email
+
+
+initial_re = re.compile(r"([^@]*)@(.*)")
+processes_re = re.compile(r"\.(.*)\+([0-9]*)@(.*)")
+
+
+def is_disabled_email_addr(email: str) -> bool:
+    """Indique si un email a été transformé
+    :param email: l'adresse mail testée
+    :return: True si l'adresse a été transformée
+    """
+    return email[0] == "."
+
+
+def disable_email_addr(email: str, rang: int) -> str:
+    """transformation de l adresse. ajoute un '.' initial et ajoute le rang
+    l'email ne doit pas avoir été déjà transformée.
+    hypothèse: l'adresse mail ne comporte qu'un seul @ (en vrai un @ peut être utilisé autrement que comme séparateur
+    si quoté comme dans '"abc@def"@dom.org').
+
+    :param email: l'adresse mail à modifier
+    :param rang: le rang à ajouter
+    :return: l'adresse mail transformée.
+    """
+    if is_disabled_email_addr(email):
+        raise ScoValueError("Tentative de traitement d'une adresse déjà traitée")
+    decomposition = initial_re.match(email)
+    if decomposition is None:  # adresse mail non conforme
+        raise ScoValueError(
+            "tentative de traitement de l'adresse email non reconnue '%s' % email"
+        )
+    return ".%s+%s@%s" % (decomposition.group(1), rang, decomposition.group(2))
+
+
+def initial_email_address(email) -> (str, int):
+    """récupération de la valeur initiale de l'adresse (transformation reciproque de transform_email_adr.
+    enlève le '.' initial et supprime la terminaison '+xxx' de la partie locale.
+    Attention: il peut y avoir un autre + dans l'adresse.
+    :param email: l'adresse après transformation
+    :return: l'adresse email initiale et le rang
+    """
+    if not is_disabled_email_addr(email):
+        return (
+            email,
+            1,
+        )  # l'adresse n'a pas été transformée. on la retourne à l'identique (avec l'indice 1)
+    decomposition = processes_re.match(email)
+    if decomposition is None:  # l'adresse a été transformée mais n'est pas reconnue
+        raise ScoValueError("Adresse mail transformée non reconnue'%s'" % email)
+    rang = int(decomposition.group(2))
+    return "%s@%s" % (decomposition.group(1), decomposition.group(3)), rang
+
 
 def send_password_reset_email(user):
     token = user.get_reset_password_token()

app/auth/routes.py

@@ -21,7 +21,7 @@ from app.auth.forms import (
 )
 from app.auth.models import Permission
 from app.auth.models import User
-from app.auth.email import send_password_reset_email
+from app.auth.email import send_password_reset_email, is_disabled_email_addr
 from app.decorators import admin_required
 from app.decorators import permission_required
 
@@ -37,9 +37,16 @@ def login():
     if form.validate_on_submit():
         user = User.query.filter_by(user_name=form.user_name.data).first()
         if user is None or not user.check_password(form.password.data):
-            current_app.logger.info("login: invalid (%s)", form.user_name.data)
-            flash(_("Nom ou mot de passe invalide"))
-            return redirect(url_for("auth.login"))
+            if user and is_disabled_email_addr(user.email):
+                current_app.logger.info(
+                    "login: compte invalidé (email doublonné) (%s)", form.user_name.data
+                )
+                flash(_("compte invalidé pour conflit d'adresse email"))
+                return redirect(url_for("auth.login"))
+            else:
+                current_app.logger.info("login: invalid (%s)", form.user_name.data)
+                flash(_("Nom ou mot de passe invalide"))
+                return redirect(url_for("auth.login"))
         login_user(user, remember=form.remember_me.data)
         current_app.logger.info("login: success (%s)", form.user_name.data)
         next_page = request.args.get("next")
@@ -84,6 +91,7 @@ def reset_password_request():
     if current_user.is_authenticated:
         return redirect(url_for("scodoc.index"))
     form = ResetPasswordRequestForm()
+    breakpoint()
     if form.validate_on_submit():
         users = User.query.filter_by(email=form.email.data).all()
         if len(users) == 1:

app/scodoc/sco_users.py

@@ -30,6 +30,7 @@
 
 # Anciennement ZScoUsers.py, fonctions de gestion des données réécrite avec flask/SQLAlchemy
 import re
+from collections import defaultdict
 
 from flask import url_for, g, request
 from flask.templating import render_template
@@ -37,6 +38,7 @@ from flask_login import current_user
 
 
 from app import db, Departement
+from app.auth.email import initial_email_address
 
 from app.auth.models import Permission
 from app.auth.models import User
@@ -66,7 +68,11 @@ def index_html(all_depts=False, with_inactives=False, format="html"):
     all_depts = int(all_depts)
     with_inactives = int(with_inactives)
 
-    H = [html_sco_header.html_sem_header("Gestion des utilisateurs")]
+    H = [
+        html_sco_header.html_sem_header(
+            "Gestion des utilisateurs", cssstyles=["css/user_list.css"]
+        )
+    ]
 
     if current_user.has_permission(Permission.ScoUsersAdmin, g.scodoc_dept):
         H.append(
@@ -109,12 +115,124 @@ def index_html(all_depts=False, with_inactives=False, format="html"):
     )
     if format != "html":
         return L
+
+    H.append(
+        stats_users(
+            dept=g.scodoc_dept, all_depts=all_depts, with_inactives=with_inactives
+        )
+    )
     H.append(L)
 
     F = html_sco_header.sco_footer()
     return "\n".join(H) + F
 
 
+LIB_NO_MAIL = "** email non renseigné **"
+
+
+def _explications(
+    nbre_emails, nbre_comptes, nbre_comptes_inactifs, nbre_comptes_desactives
+):
+    html = []
+    html.append(
+        "<p>Certains comptes partagent les même adresses emails ou 'ont pas d'adresses.<br/>"
+        " Pour garantir l'unicité des adresses nous vous proposont de désactiver certains de ces comptes. </p>"
+    )
+    html.append("<p>Nombre d'adresses' concernées: %s</p>" % nbre_emails)
+    html.append("<p>Nombre de comptes concernés: %s</p>" % nbre_comptes)
+    html.append(
+        "<p>Nombre de comptes actifs parmi les concernés: %s</p>"
+        % (nbre_comptes - nbre_comptes_inactifs)
+    )
+    html.append(
+        (
+            "<p>Nombre de comptes desactivés parmi les comptes actifs: %s"
+            % nbre_comptes_desactives
+        )
+    )
+    return html
+
+
+def _tables_anomalies(email_doublon, bilan):
+    html = [
+        "<table class='duplicates'><tbody>",
+        "<tr><th>mail initial</th><th>utilisateur</th><th width='10%'>actif</th><th>mail actuel</th><th width='15%'>action</th></tr>",
+    ]
+    nbre_comptes_inactifs = 0
+    nbre_comptes_desactives = 0
+    email_doublon.sort()
+    for initial in email_doublon:
+        bilan[initial].sort(key=lambda u: (u.active, u.last_seen), reverse=True)
+        rang = 1
+        for u in bilan[initial]:
+            # ne garde que les addresses mail des comptes de rpemier rang, actifs ayant une adresse non nulle
+            if not u.active:
+                nbre_comptes_inactifs += 1
+                keep = False
+            elif rang == 1 and u.email is not None:
+                keep = True
+                nbre_comptes_desactives += 1
+            else:
+                keep = False
+            if rang == 1:
+                row_start = "<tr'><th rowspan='%s'>%s</th>" % (
+                    len(bilan[initial]),
+                    initial,
+                )
+            else:
+                row_start = "<tr>"
+            if keep:
+                row_end = "<td class='keep'>%s</td><td class='keep' align='center'>%s</td><td class='keep'>%s</td><td align='center' class='keep'>conservé</td></tr>" % (
+                    u.user_name,
+                    "oui" if u.active else "non",
+                    u.email or "** non renseigné **",
+                )
+            else:
+                row_end = "<td class='no-keep'>%s</td><td class='no-keep' align='center'>%s</td><td class='no-keep'>%s</td><td align='center' class='no-keep'>désactivé</td></tr>" % (
+                    u.user_name,
+                    "oui" if u.active else "non",
+                    u.email or "** non renseigné **",
+                )
+            rang = rang + 1
+            html.append(row_start + row_end)
+    html.append("</tbody></table>")
+    return html, nbre_comptes_inactifs, nbre_comptes_desactives
+
+
+def stats_users(dept=None, all_depts=True, with_inactives=False):
+    H = []
+    all_users = User.query.all()
+    bilan = defaultdict(list)  # inventaire des users par email
+    rang_max = defaultdict(int)
+    for u in all_users:
+        if all_depts or u.dept == dept:
+            if with_inactives or u.active:
+                if u.email is None:
+                    initial, rang = LIB_NO_MAIL, 1
+                else:
+                    initial, rang = initial_email_address(u.email)
+                bilan[initial].append(u)
+                if rang > rang_max[initial]:
+                    rang_max[initial] = rang
+    email_doublon = [email for email in bilan if len(bilan[email]) > 1 or email is None]
+    nbre_emails = len(email_doublon)
+    nbre_comptes = sum([len(bilan[email]) for email in email_doublon])
+    if len(email_doublon) > 0:
+        table, nbre_comptes_inactifs, nbre_comptes_desactives = _tables_anomalies(
+            email_doublon, bilan
+        )
+        H += table
+        nbre_comptes_actifs = nbre_comptes - +nbre_comptes_inactifs
+        H += _explications(
+            nbre_emails=nbre_emails,
+            nbre_comptes=nbre_comptes,
+            nbre_comptes_inactifs=nbre_comptes_inactifs,
+            nbre_comptes_desactives=nbre_comptes_desactives,
+        )
+        return "<div>" + "\n".join(H) + "</div>"
+    return ""
+
+
 def list_users(
     dept,
     all_depts=False,  # tous les departements

app/static/css/user_list.css

@@ -0,0 +1,35 @@
+
+.duplicates {
+    border-style: solid;
+    border-collapse: collapse;
+}
+
+.duplicates tr:first-child th:first-child {
+    border-style: solid;
+    border-width: medium medium thin thin;
+    text-align: center;
+}
+
+.duplicates tr:first-child th {
+    background-color: lightcyan;
+    border-style: solid;
+    border-width: medium thin thin thin;
+    text-align: center;
+}
+
+.duplicates tr:not(first-child) th {
+    border-style: solid;
+    border-width: thin medium thin medium;
+    text-align: left;
+}
+
+
+.duplicates td {
+    border-style:  solid ;
+    border-width: thin;
+    border-color: black;
+}
+
+.keep { color:green; text-weight: bold;}
+
+.no-keep { color:black; }

app/views/users.py

@@ -774,6 +774,7 @@ def form_change_password(user_name=None):
     )
     if request.method == "POST" and form.cancel.data:  # cancel button clicked
         return redirect(destination)
+    breakpoint()
     if form.validate_on_submit():
         messages = []
         if form.new_password.data != "":  # change password