Send email notification when enabling or disabling MFA

README.md

@@ -138,6 +138,7 @@ This next release focus on two-factor-authentication as a measure to increase se
 * Enforce better rate limit on login, mfa, password change and API [CVE-2022-3439](https://nvd.nist.gov/vuln/detail/CVE-2022-3439) [CVE-2022-3456](https://nvd.nist.gov/vuln/detail/CVE-2022-3456)
 * Enforce 'Origin' validation [CVE-2022-3457](https://nvd.nist.gov/vuln/detail/CVE-2022-3457)
 * Define idle and absolute session timeout with agressive default to protect usage on public computer [CVE-2022-3327](https://nvd.nist.gov/vuln/detail/CVE-2022-3327)
+* Send email notification when enabling or disabling MFA [CVE-2022-3363](https://nvd.nist.gov/vuln/detail/CVE-2022-3363)
 
 Breaking changes:
 

rdiffweb/controller/page_mfa.py

@@ -105,10 +105,10 @@ def send_code(self):
                     "Multi-factor authentication is enabled for your account, but your account does not have a valid email address to send the verification code to. Check your account settings with your administrator."
                 )
             )
-        else:
-            code = cherrypy.tools.auth_mfa.generate_code()
-            body = self.app.templates.compile_template(
-                "email_mfa.html", **{"header_name": self.app.cfg.header_name, 'user': userobj, 'code': code}
-            )
-            cherrypy.engine.publish('queue_mail', to=userobj.email, subject=_("Your verification code"), message=body)
-            flash(_("A new verification code has been sent to your email."))
+            return
+        code = cherrypy.tools.auth_mfa.generate_code()
+        body = self.app.templates.compile_template(
+            "email_verification_code.html", **{"header_name": self.app.cfg.header_name, 'user': userobj, 'code': code}
+        )
+        cherrypy.engine.publish('queue_mail', to=userobj.email, subject=_("Your verification code"), message=body)
+        flash(_("A new verification code has been sent to your email."))

rdiffweb/controller/page_pref_mfa.py

@@ -126,7 +126,7 @@ def send_code(self):
             return
         code = cherrypy.tools.auth_mfa.generate_code()
         body = self.app.templates.compile_template(
-            "email_mfa.html", **{"header_name": self.app.cfg.header_name, 'user': userobj, 'code': code}
+            "email_verification_code.html", **{"header_name": self.app.cfg.header_name, 'user': userobj, 'code': code}
         )
         cherrypy.engine.publish('queue_mail', to=userobj.email, subject=_("Your verification code"), message=body)
         flash(_("A new verification code has been sent to your email."))

rdiffweb/controller/tests/test_page_prefs_mfa.py

@@ -15,7 +15,7 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-from unittest.mock import MagicMock
+from unittest.mock import ANY, MagicMock
 
 import cherrypy
 from parameterized import parameterized
@@ -34,47 +34,49 @@ def setUp(self):
         userobj = UserObject.get_user(self.USERNAME)
         userobj.email = 'admin@example.com'
         userobj.add()
+        # Register a listener on email
+        self.listener = MagicMock()
+        cherrypy.engine.subscribe('queue_mail', self.listener.queue_email, priority=50)
+
+    def tearDown(self):
+        cherrypy.engine.unsubscribe('queue_mail', self.listener.queue_email)
+        return super().tearDown()
 
     def _set_mfa(self, mfa):
         # Define mfa for user
         userobj = UserObject.get_user(self.USERNAME)
         userobj.mfa = mfa
         userobj.add()
+        # Reset mock.
+        self.listener.queue_email.reset_mock()
+        # Leave to disable mfa
         if mfa == UserObject.DISABLED_MFA:
             return
         # Generate a code for login if required
-        self.listener = MagicMock()
-        cherrypy.engine.subscribe('queue_mail', self.listener.queue_email, priority=50)
-        try:
-            self.getPage("/mfa/")
-            self.assertStatus(200)
-            self.assertInBody("A new verification code has been sent to your email.")
-            # Extract code from email between <strong> and </strong>
-            self.listener.queue_email.assert_called_once()
-            message = self.listener.queue_email.call_args[1]['message']
-            code = message.split('<strong>', 1)[1].split('</strong>')[0]
-            # Login to MFA
-            self.getPage("/mfa/", method='POST', body={'code': code, 'submit': '1'})
-            self.assertStatus(303)
-        finally:
-            cherrypy.engine.unsubscribe('queue_mail', self.listener.queue_email)
+        self.getPage("/mfa/")
+        self.assertStatus(200)
+        self.assertInBody("A new verification code has been sent to your email.")
+        # Extract code from email between <strong> and </strong>
+        self.listener.queue_email.assert_called_once()
+        message = self.listener.queue_email.call_args[1]['message']
+        code = message.split('<strong>', 1)[1].split('</strong>')[0]
+        # Login to MFA
+        self.getPage("/mfa/", method='POST', body={'code': code, 'submit': '1'})
+        self.assertStatus(303)
+        # Clear mock.
+        self.listener.queue_email.reset_mock()
 
     def _get_code(self, action):
         assert action in ['enable_mfa', 'disable_mfa', 'resend_code']
-        # Register an email listeer to capture email send
-        self.listener = MagicMock()
-        cherrypy.engine.subscribe('queue_mail', self.listener.queue_email, priority=50)
         # Query MFA page to generate a code
-        try:
-            self.getPage("/prefs/mfa", method='POST', body={action: '1'})
-            self.assertStatus(200)
-            self.assertInBody("A new verification code has been sent to your email.")
-            # Extract code from email between <strong> and </strong>
-            self.listener.queue_email.assert_called_once()
-            message = self.listener.queue_email.call_args[1]['message']
-            return message.split('<strong>', 1)[1].split('</strong>')[0]
-        finally:
-            cherrypy.engine.unsubscribe('queue_mail', self.listener.queue_email)
+        self.getPage("/prefs/mfa", method='POST', body={action: '1'})
+        self.assertStatus(200)
+        self.assertInBody("A new verification code has been sent to your email.")
+        # Extract code from email between <strong> and </strong>
+        self.listener.queue_email.assert_called_once()
+        message = self.listener.queue_email.call_args[1]['message']
+        self.listener.queue_email.reset_mock()
+        return message.split('<strong>', 1)[1].split('</strong>')[0]
 
     def test_get(self):
         # When getting the page
@@ -84,11 +86,11 @@ def test_get(self):
 
     @parameterized.expand(
         [
-            ('enable_mfa', UserObject.DISABLED_MFA, UserObject.ENABLED_MFA),
-            ('disable_mfa', UserObject.ENABLED_MFA, UserObject.DISABLED_MFA),
+            ('enable_mfa', UserObject.DISABLED_MFA, UserObject.ENABLED_MFA, "Two-Factor Authentication turned on"),
+            ('disable_mfa', UserObject.ENABLED_MFA, UserObject.DISABLED_MFA, "Two-Factor Authentication turned off"),
         ]
     )
-    def test_with_valid_code(self, action, initial_mfa, expected_mfa):
+    def test_with_valid_code(self, action, initial_mfa, expected_mfa, expected_subject):
         # Define mfa for user
         self._set_mfa(initial_mfa)
         # Given a user with email requesting a code
@@ -99,8 +101,10 @@ def test_with_valid_code(self, action, initial_mfa, expected_mfa):
         self.assertStatus(200)
         userobj = UserObject.get_user(self.USERNAME)
         self.assertEqual(userobj.mfa, expected_mfa)
-        # Then no email get sent
+        # Then no verification code get sent
         self.assertNotInBody("A new verification code has been sent to your email.")
+        # Then an email confirmation get send
+        self.listener.queue_email.assert_called_once_with(to=ANY, subject=expected_subject, message=ANY)
         # Then next page request is still working.
         self.getPage('/')
         self.assertStatus(200)

rdiffweb/core/config.py

@@ -159,7 +159,7 @@ def get_parser():
         '--emailsendchangednotification',
         help='True to send notification when sensitive information get change in user profile.',
         action='store_true',
-        default=False,
+        default=True,
     )
 
     parser.add_argument(

rdiffweb/core/model/_user.py

@@ -24,7 +24,7 @@
 from sqlalchemy import Column, Integer, SmallInteger, String, and_, event, inspect, or_
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.hybrid import hybrid_property
-from sqlalchemy.orm import deferred, relationship
+from sqlalchemy.orm import deferred, relationship, validates
 from zxcvbn import zxcvbn
 
 import rdiffweb.tools.db  # noqa
@@ -74,9 +74,9 @@ class UserObject(Base):
     PATTERN_USERNAME = r"[a-zA-Z0-9_.\-]+$"
 
     userid = Column('UserID', Integer, primary_key=True)
-    _username = Column('Username', String, nullable=False, unique=True)
+    username = Column('Username', String, nullable=False, unique=True)
     hash_password = Column('Password', String, nullable=False, default="")
-    _user_root = Column('UserRoot', String, nullable=False, default="")
+    user_root = Column('UserRoot', String, nullable=False, default="")
     _is_admin = deferred(
         Column(
             'IsAdmin',
@@ -86,7 +86,7 @@ class UserObject(Base):
             doc="DEPRECATED This column is replaced by 'role'",
         )
     )
-    _email = Column('UserEmail', String, nullable=False, default="")
+    email = Column('UserEmail', String, nullable=False, default="")
     restore_format = deferred(
         Column(
             'RestoreFormat',
@@ -96,7 +96,7 @@ class UserObject(Base):
             doc="DEPRECATED This column is not used anymore",
         )
     )
-    _role = Column('role', SmallInteger, nullable=False, server_default=str(USER_ROLE))
+    role = Column('role', SmallInteger, nullable=False, server_default=str(USER_ROLE), default=USER_ROLE)
     fullname = Column('fullname', String, nullable=False, default="")
     mfa = Column('mfa', SmallInteger, nullable=False, default=DISABLED_MFA)
     repo_objs = relationship(
@@ -129,7 +129,7 @@ def create_admin_user(cls, default_username, default_password):
         userobj.add()
 
     @classmethod
-    def add_user(cls, username, password=None, **attrs):
+    def add_user(cls, username, password=None, role=USER_ROLE, **attrs):
         """
         Used to add a new user with an optional password.
         """
@@ -143,6 +143,7 @@ def add_user(cls, username, password=None, **attrs):
         userobj = UserObject(
             username=username,
             hash_password=hash_password(password) if password else '',
+            role=role,
             **attrs,
         ).add()
         # Raise event
@@ -383,51 +384,11 @@ def set_password(self, password):
     def __eq__(self, other):
         return type(self) == type(other) and inspect(self).key == inspect(other).key
 
-    @hybrid_property
-    def username(self):
-        return self._username
-
-    @username.setter
-    def username(self, value):
-        oldvalue = self._username
-        self._username = value
-        if oldvalue != value:
-            cherrypy.engine.publish('user_attr_changed', self, {'username': (oldvalue, value)})
-
-    @hybrid_property
-    def role(self):
-        if self._role is None:
-            return self.USER_ROLE
-        return self._role
-
-    @role.setter
-    def role(self, value):
-        oldvalue = self._role
-        self._role = value
-        if oldvalue != value:
-            cherrypy.engine.publish('user_attr_changed', self, {'role': (oldvalue, value)})
-
-    @hybrid_property
-    def email(self):
-        return self._email
-
-    @email.setter
-    def email(self, value):
-        oldvalue = self._email
-        self._email = value
-        if oldvalue != value:
-            cherrypy.engine.publish('user_attr_changed', self, {'email': (oldvalue, value)})
-
-    @hybrid_property
-    def user_root(self):
-        return self._user_root
-
-    @user_root.setter
-    def user_root(self, value):
-        oldvalue = self._user_root
-        self._user_root = value
-        if oldvalue != value:
-            cherrypy.engine.publish('user_attr_changed', self, {'user_root': (oldvalue, value)})
+    @validates('username')
+    def validates_username(self, key, value):
+        if self.username:
+            raise ValueError('Username cannot be modified.')
+        return value
 
     def validate_access_token(self, token):
         """
@@ -460,3 +421,19 @@ def user_after_delete(mapper, connection, target):
     Publish event when user is deleted.
     """
     cherrypy.engine.publish('user_deleted', target.username)
+
+
+@event.listens_for(UserObject, 'after_update')
+def user_attr_changed(mapper, connection, target):
+    changes = {}
+    state = inspect(target)
+    for attr in state.attrs:
+        if attr.key in ['user_root', 'email', 'role', 'mfa']:
+            hist = attr.load_history()
+            if hist.has_changes():
+                changes[attr.key] = (
+                    hist.deleted[0] if len(hist.deleted) >= 1 else None,
+                    hist.added[0] if len(hist.added) >= 1 else None,
+                )
+    if changes:
+        cherrypy.engine.publish('user_attr_changed', target, changes)

rdiffweb/core/model/tests/test_user.py

@@ -36,11 +36,6 @@
 
 
 class UserObjectTest(rdiffweb.test.WebCase):
-
-    default_config = {
-        'email-send-changed-notification': True,
-    }
-
     def _read_ssh_key(self):
         """Readthe pub key from test packages"""
         filename = pkg_resources.resource_filename('rdiffweb.core.tests', 'test_publickey_ssh_rsa.pub')
@@ -174,12 +169,16 @@ def test_get_set(self):
         user.refresh_repos()
         self.listener.user_attr_changed.assert_called_with(user, {'user_root': ('', self.testcases)})
         self.listener.user_attr_changed.reset_mock()
+        user = UserObject.get_user('larry')
         user.role = UserObject.ADMIN_ROLE
+        user.add()
         self.listener.user_attr_changed.assert_called_with(
             user, {'role': (UserObject.USER_ROLE, UserObject.ADMIN_ROLE)}
         )
         self.listener.user_attr_changed.reset_mock()
+        user = UserObject.get_user('larry')
         user.email = 'larry@gmail.com'
+        user.add()
         self.listener.user_attr_changed.assert_called_with(user, {'email': ('', 'larry@gmail.com')})
         self.listener.user_attr_changed.reset_mock()
 

rdiffweb/core/notification.py

@@ -78,19 +78,31 @@ def user_attr_changed(self, userobj, attrs={}):
             return
 
         # Leave if the mail was not changed.
-        if 'email' not in attrs:
-            return
-
-        old_email = attrs['email'][0]
-        if not old_email:
-            logger.info("can't sent mail to user [%s] without an email", userobj.username)
-            return
-
-        # If the email attributes was changed, send a mail notification.
-        body = self.app.templates.compile_template(
-            "email_changed.html", **{"header_name": self.app.cfg.header_name, 'user': userobj}
-        )
-        self.bus.publish('queue_mail', to=old_email, subject=_("Email address changed"), message=body)
+        if 'email' in attrs:
+            old_email = attrs['email'][0]
+            if not old_email:
+                logger.info("can't sent mail to user [%s] without an email", userobj.username)
+                return
+            # If the email attributes was changed, send a mail notification.
+            subject = _("Email address changed")
+            body = self.app.templates.compile_template(
+                "email_changed.html", **{"header_name": self.app.cfg.header_name, 'user': userobj}
+            )
+            self.bus.publish('queue_mail', to=old_email, subject=str(subject), message=body)
+
+        if 'mfa' in attrs:
+            if not userobj.email:
+                logger.info("can't sent mail to user [%s] without an email", userobj.username)
+                return
+            subject = (
+                _("Two-Factor Authentication turned off")
+                if userobj.mfa == UserObject.DISABLED_MFA
+                else _("Two-Factor Authentication turned on")
+            )
+            body = self.app.templates.compile_template(
+                "email_mfa.html", **{"header_name": self.app.cfg.header_name, 'user': userobj}
+            )
+            self.bus.publish('queue_mail', to=userobj.email, subject=str(subject), message=body)
 
     def user_password_changed(self, userobj):
         if not self.send_changed:

rdiffweb/core/tests/test_notification.py

@@ -118,10 +118,13 @@ def test_email_changed(self):
         # Given a user with an email address
         user = UserObject.get_user(self.USERNAME)
         user.email = 'original_email@test.com'
+        user.add()
         self.listener.queue_email.reset_mock()
 
         # When updating the user's email
+        user = UserObject.get_user(self.USERNAME)
         user.email = 'email_changed@test.com'
+        user.add()
 
         # Then a email is queue to notify the user.
         self.listener.queue_email.assert_called_once_with(