# -*- 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 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 = [] 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"" 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"]) # 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 "".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 "".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))