Merge pull request #345 from tracim/fix/307/webdav_auth_once_pwd_email_changed

tracim/tracim/command/user.py

 
         try:
             user = User(email=login, password=password, **kwargs)
-            user.update_webdav_digest_auth(password)
             self._session.add(user)
             self._session.flush()
 
     def _update_password_for_login(self, login, password):
         user = self._user_api.get_one_by_email(login)
         user.password = password
-        user.update_webdav_digest_auth(password)
         self._session.flush()
         transaction.commit()
 

tracim/tracim/controllers/admin/user.py

 # -*- coding: utf-8 -*-
-import uuid
 import random
 
 import pytz
-from tracim import model  as pm
+from tracim import model as pm
 
-from sprox.tablebase import TableBase
-from sprox.formbase import EditableForm, AddRecordForm
-from sprox.fillerbase import TableFiller, EditFormFiller
-from tracim.config.app_cfg import CFG
-from tw2 import forms as tw2f
 import tg
 from tg import predicates
 from tg import tmpl_context
 from tg.i18n import ugettext as _
 
-from sprox.widgets import PropertyMultipleSelectField
-from sprox._compat import unicode_text
-
-from formencode import Schema
-from formencode.validators import FieldsMatch
-
 from tracim.controllers import TIMRestController
 from tracim.controllers.user import UserWorkspaceRestController
 
 from tracim.lib import helpers as h
 from tracim.lib.base import logger
 from tracim.lib.email import get_email_manager
-from tracim.lib.user import UserApi
 from tracim.lib.group import GroupApi
+from tracim.lib.user import UserApi
 from tracim.lib.userworkspace import RoleApi
 from tracim.lib.workspace import WorkspaceApi
 
 from tracim.model import DBSession
-from tracim.model.auth import Group, User
-from tracim.model.serializers import Context, CTX, DictLikeClass
+from tracim.model.auth import Group
+from tracim.model.serializers import CTX
+from tracim.model.serializers import Context
+from tracim.model.serializers import DictLikeClass
+
 
 class UserProfileAdminRestController(TIMRestController):
-    """
-     CRUD Controller allowing to manage groups of a user
-    """
+    """CRUD Controller allowing to manage groups of a user."""
 
     allow_only = predicates.in_group(Group.TIM_ADMIN_GROUPNAME)
 
     @property
     def allowed_profiles(self):
         return [
-        UserProfileAdminRestController._ALLOWED_PROFILE_USER,
-        UserProfileAdminRestController._ALLOWED_PROFILE_MANAGER,
-        UserProfileAdminRestController._ALLOWED_PROFILE_ADMIN
-    ]
+            UserProfileAdminRestController._ALLOWED_PROFILE_USER,
+            UserProfileAdminRestController._ALLOWED_PROFILE_MANAGER,
+            UserProfileAdminRestController._ALLOWED_PROFILE_ADMIN,
+        ]
 
     def _before(self, *args, **kw):
         """
-        Instantiate the current workspace in tg.tmpl_context
+        Instantiate the current workspace in tg.tmpl_context.
+
         :param args:
         :param kw:
         :return:
         tg.tmpl_context.user = user
 
     @tg.expose()
-    def switch(self, new_role):
+    def switch(self, new_role) -> None:
         """
-        :param new_role: value should be 'tracim-user', 'tracim-manager' (allowed to create workspaces) or 'tracim-admin' (admin the whole system)
-        :return:
+        Switch to the given new role.
+
+        :param new_role: value should be:
+            'tracim-user',
+            'tracim-manager' (allowed to create workspaces) or
+            'tracim-admin' (admin the whole system)
         """
         return self.put(new_role)
 
 
         group_api = GroupApi(current_user)
 
-        if current_user.user_id==user.user_id:
+        if current_user.user_id == user.user_id:
             tg.flash(_('You can\'t change your own profile'), CST.STATUS_ERROR)
             tg.redirect(self.parent_controller.url())
 
-
         redirect_url = self.parent_controller.url(skip_id=True)
 
         if new_profile not in self.allowed_profiles:
         pod_manager_group = group_api.get_one(Group.TIM_MANAGER)
         pod_admin_group = group_api.get_one(Group.TIM_ADMIN)
 
-        flash_message = _('User updated.') # this is the default value ; should never appear
+        # this is the default value ; should never appear
+        flash_message = _('User updated.')
 
-        if new_profile==UserProfileAdminRestController._ALLOWED_PROFILE_USER:
+        if new_profile == UserProfileAdminRestController._ALLOWED_PROFILE_USER:
             if pod_user_group not in user.groups:
                 user.groups.append(pod_user_group)
 
 
             flash_message = _('User {} is now a basic user').format(user.get_display_name())
 
-        elif new_profile==UserProfileAdminRestController._ALLOWED_PROFILE_MANAGER:
+        elif new_profile == UserProfileAdminRestController._ALLOWED_PROFILE_MANAGER:
             if pod_user_group not in user.groups:
                 user.groups.append(pod_user_group)
             if pod_manager_group not in user.groups:
 
             flash_message = _('User {} can now workspaces').format(user.get_display_name())
 
-
-        elif new_profile==UserProfileAdminRestController._ALLOWED_PROFILE_ADMIN:
+        elif new_profile == UserProfileAdminRestController._ALLOWED_PROFILE_ADMIN:
             if pod_user_group not in user.groups:
                 user.groups.append(pod_user_group)
             if pod_manager_group not in user.groups:
             flash_message = _('User {} is now an administrator').format(user.get_display_name())
 
         else:
-            logger.error(self, 'Trying to change user {} profile with unexpected profile {}'.format(user.user_id, new_profile))
+            error_msg = \
+                'Trying to change user {} profile with unexpected profile {}'
+            logger.error(self, error_msg.format(user.user_id, new_profile))
             tg.flash(_('Unknown profile'), CST.STATUS_ERROR)
             tg.redirect(redirect_url)
 
         pass
 
 
-
 class UserPasswordAdminRestController(TIMRestController):
-    """
-     CRUD Controller allowing to manage password of a given user
-    """
+    """CRUD Controller allowing to manage password of a given user."""
 
-    allow_only = predicates.in_any_group(Group.TIM_MANAGER_GROUPNAME, Group.TIM_ADMIN_GROUPNAME)
+    allow_only = predicates.in_any_group(
+            Group.TIM_MANAGER_GROUPNAME,
+            Group.TIM_ADMIN_GROUPNAME,
+        )
 
     def _before(self, *args, **kw):
         """
-        Instantiate the current workspace in tg.tmpl_context
+        Instantiate the current workspace in tg.tmpl_context.
+
         :param args:
         :param kw:
         :return:
         tg.tmpl_context.user_id = user_id
         tg.tmpl_context.user = user
 
-
     @tg.expose('tracim.templates.admin.user_password_edit')
     def edit(self):
         current_user = tmpl_context.current_user
         api = UserApi(current_user)
         dictified_user = Context(CTX.USER).toDict(tmpl_context.user, 'user')
-        return DictLikeClass(result = dictified_user)
+        return DictLikeClass(result=dictified_user)
 
     @tg.expose()
     def put(self, new_password1, new_password2, next_url=''):
             tg.flash(_('Empty password is not allowed.'), CST.STATUS_ERROR)
             tg.redirect(next_url)
 
-        if new_password1!=new_password2:
+        if new_password1 != new_password2:
             tg.flash(_('New passwords do not match.'), CST.STATUS_ERROR)
             tg.redirect(next_url)
 
         user.password = new_password1
-        user.update_webdav_digest_auth(new_password1)
         pm.DBSession.flush()
 
         tg.flash(_('The password has been changed'), CST.STATUS_OK)
 
     def _before(self, *args, **kw):
         """
-        Instantiate the current workspace in tg.tmpl_context
+        Instantiate the current workspace in tg.tmpl_context.
+
         :param args:
         :param kw:
         :return:
 
 
 class UserRestController(TIMRestController):
-    """
-     CRUD Controller allowing to manage Users
-    """
-    allow_only = predicates.in_any_group(Group.TIM_MANAGER_GROUPNAME, Group.TIM_ADMIN_GROUPNAME)
+    """CRUD Controller allowing to manage Users."""
+
+    allow_only = predicates.in_any_group(
+            Group.TIM_MANAGER_GROUPNAME,
+            Group.TIM_ADMIN_GROUPNAME,
+        )
 
     password = UserPasswordAdminRestController()
     profile = UserProfileAdminRestController()
     def current_item_id_key_in_context(cls):
         return 'user_id'
 
-
     @tg.require(predicates.in_group(Group.TIM_MANAGER_GROUPNAME))
     @tg.expose('tracim.templates.admin.user_getall')
     def get_all(self, *args, **kw):
         fake_api = Context(CTX.USERS).toDict({'current_user': current_user_content})
 
         dictified_users = Context(CTX.USERS).toDict(users, 'users', 'user_nb')
-        return DictLikeClass(result = dictified_users, fake_api=fake_api)
+        return DictLikeClass(result=dictified_users, fake_api=fake_api)
 
     @tg.require(predicates.in_group(Group.TIM_MANAGER_GROUPNAME))
     @tg.expose()
             password = self.generate_password()
             user.password = password
 
-        user.webdav_left_digest_response_hash = '%s:/:%s' % (email, password)
-
         api.save(user)
 
         # Now add the user to related groups
     @classmethod
     def generate_password(
             cls,
-            password_length = PASSWORD_LENGTH,
-            password_chars = PASSWORD_CHARACTERS
-            ):
-
+            password_length=PASSWORD_LENGTH,
+            password_chars=PASSWORD_CHARACTERS,
+    ):
         # character list that will be contained into the password
         char_list = []
 
-        for j in range(0, password_length):
+        for _unused in range(password_length):
             # This puts a random char from the list above inside
             # the list of chars and then merges them into a String
             char_list.append(random.choice(password_chars))
     @tg.expose('tracim.templates.admin.user_getone')
     def get_one(self, user_id):
         current_user = tmpl_context.current_user
-        api = UserApi(current_user )
+        api = UserApi(current_user)
         # role_api = RoleApi(tg.tmpl_context.current_user)
         # user_api = UserApi(tg.tmpl_context.current_user)
 
-        user = api.get_one(user_id) # FIXME
+        user = api.get_one(user_id)  # FIXME
 
         role_api = RoleApi(tg.tmpl_context.current_user)
         role_list = role_api.get_roles_for_select_field()
                                          role_types=role_list)
         fake_api = Context(CTX.ADMIN_USER).toDict(fake_api_content)
 
-        return DictLikeClass(result = dictified_user, fake_api=fake_api)
-
+        return DictLikeClass(result=dictified_user, fake_api=fake_api)
 
     @tg.expose('tracim.templates.admin.user_edit')
     def edit(self, id):
             tg.redirect(next_url)
         tg.redirect(self.url())
 
-
     @tg.require(predicates.in_group(Group.TIM_ADMIN_GROUPNAME))
     @tg.expose()
     def enable(self, id, next_url=None):
         api.save(user)
 
         tg.flash(_('User {} enabled.').format(user.get_display_name()), CST.STATUS_OK)
-        if next_url=='user':
+        if next_url == 'user':
             tg.redirect(self.url(id=user.user_id))
         tg.redirect(self.url())
 
         current_user = tmpl_context.current_user
         api = UserApi(current_user)
 
-        if current_user.user_id==id:
+        if current_user.user_id == id:
             tg.flash(_('You can\'t de-activate your own account'), CST.STATUS_ERROR)
         else:
             user = api.get_one(id)
             api.save(user)
             tg.flash(_('User {} disabled').format(user.get_display_name()), CST.STATUS_OK)
 
-        if next_url=='user':
+        if next_url == 'user':
             tg.redirect(self.url(id=user.user_id))
         tg.redirect(self.url())

tracim/tracim/controllers/user.py

             tg.redirect(redirect_url)
 
         current_user.password = new_password1
-        current_user.update_webdav_digest_auth(new_password1)
         pm.DBSession.flush()
 
         tg.flash(_('Your password has been changed'))

tracim/tracim/lib/auth/internal.py

 # -*- coding: utf-8 -*-
-from sqlalchemy import and_
+from typing import Dict
+
 from tg.configuration.auth import TGAuthMetadata
 
 from tracim.lib.auth.base import Auth
 from tracim.model import DBSession, User
 
-# TODO : temporary fix to update DB, to remove
-import transaction
 
 class InternalAuth(Auth):
 
     name = 'internal'
     _internal = True
 
-    def feed_config(self):
-        """
-        Fill config with internal (database) auth information.
-        :return:
-        """
+    def feed_config(self) -> None:
+        """Fill config with internal (database) auth information."""
         super().feed_config()
         self._config['sa_auth'].user_class = User
         self._config['auth_backend'] = 'sqlalchemy'
         self._config['sa_auth'].dbsession = DBSession
-        self._config['sa_auth'].authmetadata = InternalApplicationAuthMetadata(self._config.get('sa_auth'))
+        self._config['sa_auth'].authmetadata = \
+            InternalApplicationAuthMetadata(self._config.get('sa_auth'))
 
 
 class InternalApplicationAuthMetadata(TGAuthMetadata):
+
     def __init__(self, sa_auth):
         self.sa_auth = sa_auth
 
-    def authenticate(self, environ, identity, allow_auth_token: bool=False):
-        user = self.sa_auth.dbsession.query(self.sa_auth.user_class).filter(and_(
-            self.sa_auth.user_class.is_active == True,
-            self.sa_auth.user_class.email == identity['login']
-        )).first()
-
-        if user and user.validate_password(identity['password']):
-            if not user.webdav_left_digest_response_hash:
-                user.webdav_left_digest_response_hash = '%s:/:%s' % (identity['login'], identity['password'])
-                DBSession.flush()
-                # TODO : temporary fix to update DB, to remove
-                transaction.commit()
-            return identity['login']
-
-        if user and allow_auth_token:
-            user.ensure_auth_token()
-            if user.auth_token == identity['password']:
-                return identity['login']
+    def authenticate(
+            self,
+            environ: Dict[str, str],
+            identity: Dict[str, str],
+            allow_auth_token: bool = False,
+    ) -> str:
+        """
+        Authenticate using given credentials.
+
+        Checks password first then auth token if allowed.
+        :param environ:
+        :param identity: The given credentials to authenticate.
+        :param allow_auth_token: The indicator of auth token use.
+        :return: The given login or an empty string if auth failed.
+        """
+        result = ''
+        user = self.sa_auth.dbsession \
+            .query(self.sa_auth.user_class) \
+            .filter(self.sa_auth.user_class.is_active.is_(True)) \
+            .filter(self.sa_auth.user_class.email == identity['login']) \
+            .first()
+        if user:
+            if user.validate_password(identity['password']):
+                result = identity['login']
+            if allow_auth_token:
+                user.ensure_auth_token()
+                if user.auth_token == identity['password']:
+                    result = identity['login']
+        return result
 
     def get_user(self, identity, userid):
-        return self.sa_auth.dbsession.query(self.sa_auth.user_class).filter(
-            and_(self.sa_auth.user_class.is_active == True, self.sa_auth.user_class.email == userid)).first()
+        return self.sa_auth.dbsession \
+            .query(self.sa_auth.user_class) \
+            .filter(self.sa_auth.user_class.is_active.is_(True)) \
+            .filter(self.sa_auth.user_class.email == userid) \
+            .first()
 
     def get_groups(self, identity, userid):
         return [g.group_name for g in identity['user'].groups]

tracim/tracim/model/auth.py

 
 It's perfectly fine to re-use this definition in the tracim application,
 though.
-
 """
+import os
+import time
 import uuid
 
-import os
 from datetime import datetime
-import time
-from hashlib import sha256
-from sqlalchemy.ext.hybrid import hybrid_property
-from tracim.lib.utils import lazy_ugettext as l_
 from hashlib import md5
-
-__all__ = ['User', 'Group', 'Permission']
+from hashlib import sha256
+from typing import TYPE_CHECKING
 
 from sqlalchemy import Column
 from sqlalchemy import ForeignKey
 from sqlalchemy import Sequence
 from sqlalchemy import Table
-
-from sqlalchemy.types import Unicode
-from sqlalchemy.types import Integer
-from sqlalchemy.types import DateTime
+from sqlalchemy.ext.hybrid import hybrid_property
+from sqlalchemy.orm import relation
+from sqlalchemy.orm import relationship
+from sqlalchemy.orm import synonym
 from sqlalchemy.types import Boolean
-from sqlalchemy.orm import relation, relationship, synonym
-from tg import request
-from tracim.model import DeclarativeBase, metadata, DBSession
+from sqlalchemy.types import DateTime
+from sqlalchemy.types import Integer
+from sqlalchemy.types import Unicode
+
+from tracim.lib.utils import lazy_ugettext as l_
+from tracim.model import DBSession
+from tracim.model import DeclarativeBase
+from tracim.model import metadata
+if TYPE_CHECKING:
+    from tracim.model.data import Workspace
+
+__all__ = ['User', 'Group', 'Permission']
 
 # This is the association table for the many-to-many relationship between
 # groups and permissions.
         onupdate="CASCADE", ondelete="CASCADE"), primary_key=True)
 )
 
+
 class Group(DeclarativeBase):
 
     TIM_NOBODY = 0
         return DBSession.query(cls).filter_by(group_name=group_name).first()
 
 
-
 class Profile(object):
-    """ This model is the "max" group associated to a given user
-    """
+    """This model is the "max" group associated to a given user."""
 
     _NAME = [Group.TIM_NOBODY_GROUPNAME,
              Group.TIM_USER_GROUPNAME,
         self.label = Profile._LABEL[profile_id]
 
 
-
 class User(DeclarativeBase):
     """
     User definition.
 
     This is the user definition used by :mod:`repoze.who`, which requires at
     least the ``email`` column.
-
     """
+
     __tablename__ = 'users'
 
     user_id = Column(Integer, Sequence('seq__users__user_id'), autoincrement=True, primary_key=True)
     @property
     def profile(self) -> Profile:
         profile_id = 0
-        if len(self.groups)>0:
+        if len(self.groups) > 0:
             profile_id = max(group.group_id for group in self.groups)
         return Profile(profile_id)
 
         return DBSession.query(cls).filter_by(email=username).first()
 
     @classmethod
-    def _hash_password(cls, password):
+    def _hash_password(cls, cleartext_password: str) -> str:
         salt = sha256()
         salt.update(os.urandom(60))
         salt = salt.hexdigest()
 
         hash = sha256()
         # Make sure password is a str because we cannot hash unicode objects
-        hash.update((password + salt).encode('utf-8'))
+        hash.update((cleartext_password + salt).encode('utf-8'))
         hash = hash.hexdigest()
 
-        password = salt + hash
+        ciphertext_password = salt + hash
 
         # Make sure the hashed password is a unicode object at the end of the
         # process because SQLAlchemy _wants_ unicode objects for Unicode cols
         # FIXME - D.A. - 2013-11-20 - The following line has been removed since using python3. Is this normal ?!
         # password = password.decode('utf-8')
 
-        return password
+        return ciphertext_password
 
-    def _set_password(self, password):
-        """Hash ``password`` on the fly and store its hashed version."""
-        self._password = self._hash_password(password)
+    def _set_password(self, cleartext_password: str) -> None:
+        """
+        Set ciphertext password from cleartext password.
 
-    def _get_password(self):
+        Hash cleartext password on the fly,
+        Store its ciphertext version,
+        Update the WebDAV hash as well.
+        """
+        self._password = self._hash_password(cleartext_password)
+        self.update_webdav_digest_auth(cleartext_password)
+
+    def _get_password(self) -> str:
         """Return the hashed version of the password."""
         return self._password
 
 
     webdav_left_digest_response_hash = synonym('_webdav_left_digest_response_hash',
                                                descriptor=property(_get_hash_digest,
-                                                                    _set_hash_digest))
+                                                                   _set_hash_digest))
 
-    def update_webdav_digest_auth(self, password) -> None:
+    def update_webdav_digest_auth(self, cleartext_password: str) -> None:
         self.webdav_left_digest_response_hash \
-            = '{username}:/:{password}'.format(
+            = '{username}:/:{cleartext_password}'.format(
                 username=self.email,
-                password=password,
+                cleartext_password=cleartext_password,
             )
 
-
-    def validate_password(self, password):
+    def validate_password(self, cleartext_password: str) -> bool:
         """
         Check the password against existing credentials.
 
-        :param password: the password that was provided by the user to
-            try and authenticate. This is the clear text version that we will
-            need to match against the hashed one in the database.
-        :type password: unicode object.
+        :param cleartext_password: the password that was provided by the user
+            to try and authenticate. This is the clear text version that we
+            will need to match against the hashed one in the database.
+        :type cleartext_password: unicode object.
         :return: Whether the password is valid.
         :rtype: bool
 
         """
-        if not self.password:
-            return False
-        hash = sha256()
-        hash.update((password + self.password[:64]).encode('utf-8'))
-        return self.password[64:] == hash.hexdigest()
-
-    def get_display_name(self, remove_email_part=False):
+        result = False
+        if self.password:
+            hash = sha256()
+            hash.update((cleartext_password + self.password[:64]).encode('utf-8'))
+            result = self.password[64:] == hash.hexdigest()
+            if result and not self.webdav_left_digest_response_hash:
+                self.update_webdav_digest_auth(cleartext_password)
+        return result
+
+    def get_display_name(self, remove_email_part: bool=False) -> str:
         """
+        Get a name to display from corresponding member or email.
+
         :param remove_email_part: If True and display name based on email,
-         remove @xxx.xxx part of email in returned value
+            remove @xxx.xxx part of email in returned value
         :return: display name based on user name or email.
         """
-        if self.display_name != None and self.display_name != '':
+        if self.display_name:
             return self.display_name
         else:
             if remove_email_part:
     def ensure_auth_token(self) -> None:
         """
         Create auth_token if None, regenerate auth_token if too much old.
+
         auth_token validity is set in
         :return:
         """
 
     __tablename__ = 'permissions'
 
-
     permission_id = Column(Integer, Sequence('seq__permissions__permission_id'), autoincrement=True, primary_key=True)
     permission_name = Column(Unicode(63), unique=True, nullable=False)
     description = Column(Unicode(255))
 
     def __unicode__(self):
         return self.permission_name
-