ScoDoc/app/auth/models.py

437 lines
15 KiB
Python

# -*- coding: UTF-8 -*
"""Users and Roles models for ScoDoc
"""
import base64
from datetime import datetime, timedelta
import json
import os
import re
from time import time
from typing import Optional
from flask import current_app, url_for, g
from flask_login import UserMixin, AnonymousUserMixin
from werkzeug.security import generate_password_hash, check_password_hash
import jwt
from app import db, login
from app.scodoc.sco_exceptions import ScoValueError
from app.scodoc.sco_permissions import Permission
from app.scodoc.sco_roles_default import SCO_ROLES_DEFAULTS
import app.scodoc.sco_utils as scu
from app.scodoc import sco_etud # a deplacer dans scu
VALID_LOGIN_EXP = re.compile(r"^[a-zA-Z0-9@\\\-_\\\.]+$")
class User(UserMixin, db.Model):
"""ScoDoc users, handled by Flask / SQLAlchemy"""
id = db.Column(db.Integer, primary_key=True)
user_name = db.Column(db.String(64), index=True, unique=True)
email = db.Column(db.String(120))
nom = db.Column(db.String(64))
prenom = db.Column(db.String(64))
dept = db.Column(db.String(32), index=True)
active = db.Column(db.Boolean, default=True, index=True)
password_hash = db.Column(db.String(128))
password_scodoc7 = db.Column(db.String(42))
last_seen = db.Column(db.DateTime, default=datetime.utcnow)
date_modif_passwd = db.Column(db.DateTime, default=datetime.utcnow)
date_created = db.Column(db.DateTime, default=datetime.utcnow)
date_expiration = db.Column(db.DateTime, default=None)
passwd_temp = db.Column(db.Boolean, default=False)
token = db.Column(db.String(32), index=True, unique=True)
token_expiration = db.Column(db.DateTime)
roles = db.relationship("Role", secondary="user_role", viewonly=True)
Permission = Permission
def __init__(self, **kwargs):
self.roles = []
self.user_roles = []
# check login:
if kwargs.get("user_name") and not VALID_LOGIN_EXP.match(kwargs["user_name"]):
raise ValueError(f"invalid user_name: {kwargs['user_name']}")
super(User, self).__init__(**kwargs)
# Ajoute roles:
if (
not self.roles
and self.email
and self.email == current_app.config["SCODOC_ADMIN_MAIL"]
):
# super-admin
admin_role = Role.query.filter_by(name="SuperAdmin").first()
assert admin_role
self.add_role(admin_role, None)
db.session.commit()
# current_app.logger.info("creating user with roles={}".format(self.roles))
def __repr__(self):
return f"<User {self.user_name} id={self.id} dept={self.dept}{' (inactive)' if not self.active else ''}>"
def __str__(self):
return self.user_name
def set_password(self, password):
"Set password"
current_app.logger.info(f"set_password({self})")
if password:
self.password_hash = generate_password_hash(password)
else:
self.password_hash = None
def check_password(self, password):
"""Check given password vs current one.
Returns `True` if the password matched, `False` otherwise.
"""
if not self.active: # inactived users can't login
return False
if (not self.password_hash) and self.password_scodoc7:
# Special case: user freshly migrated from ScoDoc7
if scu.check_scodoc7_password(self.password_scodoc7, password):
current_app.logger.warning(
f"migrating legacy ScoDoc7 password for {self}"
)
self.set_password(password)
self.password_scodoc7 = None
db.session.add(self)
db.session.commit()
return True
return False
if not self.password_hash: # user without password can't login
return False
return check_password_hash(self.password_hash, password)
def get_reset_password_token(self, expires_in=600):
return jwt.encode(
{"reset_password": self.id, "exp": time() + expires_in},
current_app.config["SECRET_KEY"],
algorithm="HS256",
).decode("utf-8")
@staticmethod
def verify_reset_password_token(token):
try:
id = jwt.decode(
token, current_app.config["SECRET_KEY"], algorithms=["HS256"]
)["reset_password"]
except:
return
return User.query.get(id)
def to_dict(self, include_email=True):
data = {
"date_expiration": self.date_expiration.isoformat() + "Z"
if self.date_expiration
else "",
"date_modif_passwd": self.date_modif_passwd.isoformat() + "Z"
if self.date_modif_passwd
else "",
"date_created": self.date_created.isoformat() + "Z"
if self.date_created
else "",
"dept": (self.dept or ""), # sco8
"id": self.id,
"active": self.active,
"status_txt": "actif" if self.active else "fermé",
"last_seen": self.last_seen.isoformat() + "Z",
"nom": (self.nom or ""), # sco8
"prenom": (self.prenom or ""), # sco8
"roles_string": self.get_roles_string(), # eg "Ens_RT, Ens_Info"
"user_name": self.user_name, # sco8
# Les champs calculés:
"nom_fmt": self.get_nom_fmt(),
"prenom_fmt": self.get_prenom_fmt(),
"nomprenom": self.get_nomprenom(),
"prenomnom": self.get_prenomnom(),
"nomplogin": self.get_nomplogin(),
"nomcomplet": self.get_nomcomplet(),
}
if include_email:
data["email"] = self.email or ""
return data
def from_dict(self, data, new_user=False):
"""Set users' attributes from given dict values.
Roles must be encoded as "roles_string", like "Ens_RT, Secr_CJ"
"""
for field in ["nom", "prenom", "dept", "active", "email", "date_expiration"]:
if field in data:
setattr(self, field, data[field] or None)
if new_user:
if "user_name" in data:
# never change name of existing users
self.user_name = data["user_name"]
if "password" in data:
self.set_password(data["password"])
if not VALID_LOGIN_EXP.match(self.user_name):
raise ValueError(f"invalid user_name: {self.user_name}")
# Roles: roles_string is "Ens_RT, Secr_RT, ..."
if "roles_string" in data:
self.user_roles = []
for r_d in data["roles_string"].split(","):
role, dept = UserRole.role_dept_from_string(r_d)
self.add_role(role, dept)
def get_token(self, expires_in=3600):
now = datetime.utcnow()
if self.token and self.token_expiration > now + timedelta(seconds=60):
return self.token
self.token = base64.b64encode(os.urandom(24)).decode("utf-8")
self.token_expiration = now + timedelta(seconds=expires_in)
db.session.add(self)
return self.token
def revoke_token(self):
self.token_expiration = datetime.utcnow() - timedelta(seconds=1)
@staticmethod
def check_token(token):
user = User.query.filter_by(token=token).first()
if user is None or user.token_expiration < datetime.utcnow():
return None
return user
# Permissions management:
def has_permission(self, perm: int, dept=False):
"""Check if user has permission `perm` in given `dept`.
Similar to Zope ScoDoc7 `has_permission``
Args:
perm: integer, one of the value defined in Permission class.
dept: dept id (eg 'RT'), default to current departement.
"""
if not self.active:
return False
if dept is False:
dept = g.scodoc_dept
# les role liés à ce département, et les roles avec dept=None (super-admin)
roles_in_dept = (
UserRole.query.filter_by(user_id=self.id)
.filter((UserRole.dept == dept) | (UserRole.dept == None))
.all()
)
for user_role in roles_in_dept:
if user_role.role.has_permission(perm):
return True
return False
# Role management
def add_role(self, role, dept):
"""Add a role to this user.
:param role: Role to add.
"""
self.user_roles.append(UserRole(user=self, role=role, dept=dept))
def add_roles(self, roles, dept):
"""Add roles to this user.
:param roles: Roles to add.
"""
for role in roles:
self.add_role(role, dept)
def set_roles(self, roles, dept):
"set roles in the given dept"
self.user_roles = [UserRole(user=self, role=r, dept=dept) for r in roles]
def get_roles(self):
"iterator on my roles"
for role in self.roles:
yield role
def get_roles_string(self):
"""string repr. of user's roles (with depts)
e.g. "Ens_RT, Ens_Info, Secr_CJ"
"""
return ",".join("{r.role.name}_{r.dept}".format(r=r) for r in self.user_roles)
def is_administrator(self):
"True if i'm an active SuperAdmin"
return self.active and self.has_permission(Permission.ScoSuperAdmin, dept=None)
# Some useful strings:
def get_nomplogin(self):
"""nomplogin est le nom en majuscules suivi du prénom et du login
e.g. Dupont Pierre (dupont)
"""
if self.nom:
n = sco_etud.format_nom(self.nom)
else:
n = self.user_name.upper()
return "%s %s (%s)" % (
n,
sco_etud.format_prenom(self.prenom),
self.user_name,
)
@staticmethod
def get_user_id_from_nomplogin(nomplogin: str) -> Optional[int]:
"""Returns id from the string "Dupont Pierre (dupont)"
or None if user does not exist
"""
m = re.match(r".*\((.*)\)", nomplogin.strip())
if m:
user_name = m.group(1)
u = User.query.filter_by(user_name=user_name).first()
if u:
return u.id
return None
def get_nom_fmt(self):
"""Nom formatté: "Martin" """
if self.nom:
return sco_etud.format_nom(self.nom, uppercase=False)
else:
return self.user_name
def get_prenom_fmt(self):
"""Prénom formaté (minuscule capitalisées)"""
return sco_etud.format_prenom(self.prenom)
def get_nomprenom(self):
"""Nom capitalisé suivi de l'initiale du prénom:
Viennet E.
"""
prenom_abbrv = scu.abbrev_prenom(sco_etud.format_prenom(self.prenom))
return (self.get_nom_fmt() + " " + prenom_abbrv).strip()
def get_prenomnom(self):
"""L'initiale du prénom suivie du nom: "J.-C. Dupont" """
prenom_abbrv = scu.abbrev_prenom(sco_etud.format_prenom(self.prenom))
return (prenom_abbrv + " " + self.get_nom_fmt()).strip()
def get_nomcomplet(self):
"Prénom et nom complets"
return sco_etud.format_prenom(self.prenom) + " " + self.get_nom_fmt()
# nomnoacc était le nom en minuscules sans accents (inutile)
class AnonymousUser(AnonymousUserMixin):
def has_permission(self, perm, dept=None):
return False
def is_administrator(self):
return False
login.anonymous_user = AnonymousUser
class Role(db.Model):
"""Roles for ScoDoc"""
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(64), unique=True)
default = db.Column(db.Boolean, default=False, index=True)
permissions = db.Column(db.BigInteger) # 64 bits
users = db.relationship("User", secondary="user_role", viewonly=True)
# __table_args__ = (db.UniqueConstraint("name", "dept", name="_rolename_dept_uc"),)
def __init__(self, **kwargs):
super(Role, self).__init__(**kwargs)
if self.permissions is None:
self.permissions = 0
def __repr__(self):
return "<Role {} perm={:0{w}b}>".format(
self.name,
self.permissions & ((1 << Permission.NBITS) - 1),
w=Permission.NBITS,
)
def add_permission(self, perm):
self.permissions |= perm
def remove_permission(self, perm):
self.permissions = self.permissions & ~perm
def reset_permissions(self):
self.permissions = 0
def has_permission(self, perm):
return self.permissions & perm == perm
@staticmethod
def insert_roles():
"""Create default roles"""
default_role = "Observateur"
for role_name, permissions in SCO_ROLES_DEFAULTS.items():
role = Role.query.filter_by(name=role_name).first()
if role is None:
role = Role(name=role_name)
role.reset_permissions()
for perm in permissions:
role.add_permission(perm)
role.default = role.name == default_role
db.session.add(role)
db.session.commit()
@staticmethod
def get_named_role(name):
"""Returns existing role with given name, or None."""
return Role.query.filter_by(name=name).first()
class UserRole(db.Model):
"""Associate user to role, in a dept.
If dept is None, the role applies to all departments (eg super admin).
"""
id = db.Column(db.Integer, primary_key=True)
user_id = db.Column(db.Integer, db.ForeignKey("user.id"))
role_id = db.Column(db.Integer, db.ForeignKey("role.id"))
dept = db.Column(db.String(64)) # dept acronym
user = db.relationship(
User, backref=db.backref("user_roles", cascade="all, delete-orphan")
)
role = db.relationship(
Role, backref=db.backref("user_roles", cascade="all, delete-orphan")
)
def __repr__(self):
return "<UserRole u={} r={} dept={}>".format(self.user, self.role, self.dept)
@staticmethod
def role_dept_from_string(role_dept):
"""Return tuple (role, dept) from the string
role_dept, of the forme "Role_Dept".
role is a Role instance, dept is a string.
"""
fields = role_dept.split("_", 1) # maxsplit=1, le dept peut contenir un "_"
if len(fields) != 2:
raise ScoValueError("Invalid role_dept")
role_name, dept = fields
role = Role.query.filter_by(name=role_name).first()
if role is None:
raise ScoValueError("role %s does not exists" % role_name)
return (role, dept)
def get_super_admin():
"""L'utilisateur admin (où le premier, s'il y en a plusieurs).
Utilisé par les tests unitaires et le script de migration.
"""
admin_role = Role.query.filter_by(name="SuperAdmin").first()
assert admin_role
admin_user = (
User.query.join(UserRole)
.filter((UserRole.user_id == User.id) & (UserRole.role_id == admin_role.id))
.first()
)
assert admin_user
return admin_user
@login.user_loader
def load_user(id):
return User.query.get(int(id))