Initial commit
This commit is contained in:
+196
@@ -0,0 +1,196 @@
|
||||
from datetime import datetime
|
||||
from flask import Flask, Blueprint, flash, render_template, request, url_for
|
||||
import ldap, ldap.filter, os
|
||||
from app.exceptions import LoginNotFoundException, MultipleUsersFoundForLoginException
|
||||
from flask_migrate import Migrate
|
||||
from flask_mail import Mail, Message
|
||||
from app.models import db, MagicToken
|
||||
from dotenv import load_dotenv
|
||||
load_dotenv()
|
||||
|
||||
f_app = Flask(__name__)
|
||||
f_app.secret_key = os.getenv("SECRET_KEY")
|
||||
|
||||
bp = Blueprint('mdp', __name__)
|
||||
|
||||
f_app.config["SQLALCHEMY_DATABASE_URI"] = os.getenv("DATABASE_URI")
|
||||
db.init_app(f_app)
|
||||
migrate = Migrate(f_app, db)
|
||||
|
||||
def to_bool(value, default: bool = False):
|
||||
"""
|
||||
Convert a value to a boolean
|
||||
"""
|
||||
if value is None:
|
||||
return default
|
||||
if isinstance(value, str):
|
||||
if value.lower() in ['true', '1', 't', 'y', 'yes']:
|
||||
return True
|
||||
if value.lower() in ['false', '0', 'f', 'n', 'no']:
|
||||
return False
|
||||
if value:
|
||||
return True
|
||||
return default
|
||||
|
||||
f_app.config['MAIL_SERVER']= os.getenv("MAIL_SERVER")
|
||||
f_app.config['MAIL_PORT'] = int(os.getenv("MAIL_PORT"))
|
||||
f_app.config['MAIL_USERNAME'] = os.getenv("MAIL_USERNAME")
|
||||
f_app.config['MAIL_PASSWORD'] = os.getenv("MAIL_PASSWORD")
|
||||
f_app.config['MAIL_USE_TLS'] = to_bool(os.getenv("MAIL_USE_TLS"))
|
||||
f_app.config['MAIL_USE_SSL'] = to_bool(os.getenv("MAIL_USE_SSL"))
|
||||
|
||||
mail = Mail(f_app)
|
||||
|
||||
@bp.route('/')
|
||||
def home():
|
||||
return render_template('pages/home.html.j2')
|
||||
|
||||
@bp.route('/identifiant')
|
||||
def lost_login():
|
||||
return render_template('pages/lost_login.html.j2')
|
||||
|
||||
@bp.route('/mdp', methods=("GET", "POST"))
|
||||
def password_link_form():
|
||||
if request.method == 'GET':
|
||||
return render_template('pages/password_link_form.html.j2')
|
||||
|
||||
login = request.form['login']
|
||||
error = None
|
||||
|
||||
if not login:
|
||||
error = "Le champ 'identifiant' ne peut pas être vide."
|
||||
else:
|
||||
try:
|
||||
user_dn, user_emails = ldap_get_dn_and_emails(login)
|
||||
if len(user_emails) == 0:
|
||||
error = "Aucune adresse email trouvée pour cet identifiant. Rapprochez-vous du secrétariat de votre formation pour mettre à jour votre dossier ou " + render_template("partials/contact_link.html.j2") + "."
|
||||
except LoginNotFoundException:
|
||||
error = "Aucun utilisateur trouvé avec cet identifiant.<br>Si vous avez perdu votre identifiant, <a class=\"text-blue-600 hover:underline\" href=\"" + url_for("mdp.lost_login") + "\">cliquez ici</a>."
|
||||
except MultipleUsersFoundForLoginException:
|
||||
error = "Plusieurs utilisateurs trouvés avec cet identifiant (cette erreur ne devrait jamais s'afficher, " + render_template("partials/contact_link.html.j2") + ")."
|
||||
except (ldap.SERVER_DOWN, ldap.BUSY, ldap.CONNECT_ERROR, ldap.UNAVAILABLE):
|
||||
error = "La connexion au serveur LDAP a échouée."
|
||||
except ldap.TIMEOUT:
|
||||
error = "Le serveur LDAP a mis trop de temps à répondre."
|
||||
except ldap.LDAPError:
|
||||
error = "Une erreur est survenue lors de la connexion à l'annuaire LDAP."
|
||||
|
||||
if error is not None:
|
||||
flash(error)
|
||||
return render_template('pages/password_link_form.html.j2')
|
||||
|
||||
try:
|
||||
magictoken = MagicToken.generate(user_emails, user_dn)
|
||||
db.session.add(magictoken)
|
||||
db.session.commit()
|
||||
except:
|
||||
flash("Une erreur est survenue lors de la création du lien.")
|
||||
return render_template('pages/password_link_form.html.j2')
|
||||
|
||||
try:
|
||||
send_email(user_emails, magictoken)
|
||||
anonymized_emails = [anonymize_email(email) for email in user_emails]
|
||||
except:
|
||||
flash("Une erreur est survenue lors de l'envoi du mail.")
|
||||
return render_template('pages/password_link_form.html.j2')
|
||||
|
||||
return render_template('pages/password_link_sent.html.j2', emails=anonymized_emails, is_student=True)
|
||||
|
||||
@bp.route('/mdp/<token>', methods=("GET", "POST"))
|
||||
def password_reset_form(token: str = None):
|
||||
magictoken = MagicToken.query.filter_by(token=token).first()
|
||||
if magictoken is None:
|
||||
return render_template('pages/invalid_password_token.html.j2')
|
||||
if magictoken.used:
|
||||
return render_template('pages/invalid_password_token.html.j2', used=True)
|
||||
if magictoken.expiration_date < datetime.now():
|
||||
return render_template('pages/invalid_password_token.html.j2', expired=True)
|
||||
|
||||
if request.method == 'GET':
|
||||
return render_template('pages/password_reset_form.html.j2')
|
||||
|
||||
newpassword = request.form['newpasswd']
|
||||
error = None
|
||||
|
||||
if not newpassword:
|
||||
error = "Le champ 'mot de passe' ne peut pas être vide."
|
||||
|
||||
if error is not None:
|
||||
flash(error)
|
||||
return render_template('pages/password_reset_form.html.j2')
|
||||
|
||||
try:
|
||||
ldap_set_password(magictoken.user_dn, newpassword)
|
||||
except (ldap.SERVER_DOWN, ldap.BUSY, ldap.CONNECT_ERROR, ldap.UNAVAILABLE):
|
||||
error = "La connexion au serveur LDAP a échouée."
|
||||
except ldap.TIMEOUT:
|
||||
error = "Le serveur LDAP a mis trop de temps à répondre."
|
||||
except ldap.LDAPError as err:
|
||||
error = "Erreur LDAP. " + str(err)
|
||||
|
||||
if error is not None:
|
||||
flash(error)
|
||||
return render_template('pages/password_reset_form.html.j2')
|
||||
|
||||
magictoken.used = True
|
||||
db.session.commit()
|
||||
|
||||
return render_template('pages/password_reset_success.html.j2')
|
||||
|
||||
|
||||
def send_email(emails: list[str], magictoken: MagicToken):
|
||||
"""
|
||||
Send the magic token to the list of emails
|
||||
"""
|
||||
msg = Message(subject='Initialisation du mot de passe', sender=os.getenv("MAIL_DEFAULT_SENDER"), recipients=emails)
|
||||
msg.body = render_template('emails/reset-link.txt.j2', magictoken=magictoken)
|
||||
msg.html = render_template('emails/reset-link.html.j2', magictoken=magictoken)
|
||||
mail.send(msg)
|
||||
|
||||
def anonymize_email(email: str):
|
||||
"""
|
||||
Anonymize an email address before displaying it on the website
|
||||
"""
|
||||
email_parts = email.split('@')
|
||||
email_parts[0] = email_parts[0][:3] + '*' * (len(email_parts[0]) - 3)
|
||||
known_domains = ['etu.u-pec.fr', 'u-pec.fr', 'iutsf.org', 'iut-fbleau.fr']
|
||||
if email_parts[1].lower() not in known_domains:
|
||||
email_parts[1] = email_parts[1].rsplit('.', 1)
|
||||
email_parts[1][0] = email_parts[1][0][:3] + '*' * (len(email_parts[1][0]) - 3)
|
||||
email_parts[1] = '.'.join(email_parts[1])
|
||||
|
||||
return '@'.join(email_parts)
|
||||
|
||||
def ldap_get_dn_and_emails(login: str):
|
||||
"""
|
||||
Get the user's distinguished name and email addresses from the LDAP server
|
||||
"""
|
||||
if not to_bool(os.getenv("LDAP_TLS_VERIFY")):
|
||||
ldap.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, ldap.OPT_X_TLS_NEVER)
|
||||
conn = ldap.initialize(os.getenv("LDAP_SERVER"))
|
||||
conn.simple_bind_s(os.getenv("LDAP_USER"), os.getenv("LDAP_PASSWORD"))
|
||||
result = conn.search_s(
|
||||
os.getenv("LDAP_BASE_DN"),
|
||||
ldap.SCOPE_SUBTREE,
|
||||
os.getenv("LDAP_SEARCH_FILTER") % ldap.filter.escape_filter_chars(login),
|
||||
[os.getenv("LDAP_EMAIL_ATTRIBUTE")]
|
||||
)
|
||||
if len(result) == 0:
|
||||
raise LoginNotFoundException()
|
||||
if len(result) > 1:
|
||||
raise MultipleUsersFoundForLoginException()
|
||||
return result[0][0], [entry[1][os.getenv("LDAP_EMAIL_ATTRIBUTE")][0].decode('utf-8') for entry in result]
|
||||
|
||||
def ldap_set_password(user_dn: str, newpassword: str):
|
||||
"""
|
||||
Set the user's password in LDAP
|
||||
"""
|
||||
if not to_bool(os.getenv("LDAP_TLS_VERIFY")):
|
||||
ldap.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, ldap.OPT_X_TLS_NEVER)
|
||||
conn = ldap.initialize(os.getenv("LDAP_SERVER"))
|
||||
conn.simple_bind_s(os.getenv("LDAP_USER"), os.getenv("LDAP_PASSWORD"))
|
||||
# https://serverfault.com/questions/937330/update-ad-password-from-python-ldap-code-insuff-access-rights
|
||||
conn.modify_s(user_dn, [(ldap.MOD_REPLACE, 'unicodePwd', ['"{0}"'.format(newpassword).encode('utf-16-le')])])
|
||||
|
||||
|
||||
f_app.register_blueprint(bp)
|
||||
Reference in New Issue
Block a user