Add CSRF to user and ssh management

rdiffweb/controller/cherrypy_wtf.py

@@ -1,5 +1,26 @@
-from wtforms.form import Form
+# -*- coding: utf-8 -*-
+# rdiffweb, A web interface to rdiff-backup repositories
+# Copyright (C) 2012-2021 rdiffweb contributors
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import os
+from datetime import timedelta
+
 import cherrypy
+from wtforms.csrf.session import SessionCSRF
+from wtforms.form import Form
 
 SUBMIT_METHODS = {'POST', 'PUT', 'PATCH', 'DELETE'}
 
@@ -26,14 +47,26 @@ def getlist(self, key):
 
 _AUTO = _ProxyFormdata()
 
+# Provide a default secret
+_CSRF_SECRET = os.environ.get('RDIFFWEB_CSRF_SECRET', 'b5WPUOw3Zd7gr1ligLilAAUA9zuWkW41')
+
 
 class CherryForm(Form):
     """
     Custom implementation of WTForm for cherrypy to support kwargs parms.
-    
+
     If ``formdata`` is not specified, this will use cherrypy.request.params
     Explicitly pass ``formdata=None`` to prevent this.
     """
+    class Meta:
+        csrf = True
+        csrf_class = SessionCSRF
+        csrf_secret = _CSRF_SECRET.encode('utf-8')
+        csrf_time_limit = timedelta(minutes=30)
+
+        @property
+        def csrf_context(self):
+            return cherrypy.session
 
     def __init__(self, formdata=_AUTO, **kwargs):
         super().__init__(formdata=formdata, **kwargs)
@@ -51,3 +84,8 @@ def validate_on_submit(self):
         This is a shortcut for ``form.is_submitted() and form.validate()``.
         """
         return self.is_submitted() and self.validate()
+
+    @property
+    def error_message(self):
+        if self.errors:
+            return ' '.join(['%s: %s' % (field, ', '.join(messages)) for field, messages in self.errors.items()])

rdiffweb/controller/page_admin.py

@@ -22,22 +22,21 @@
 import pwd
 import subprocess
 import sys
+from collections import OrderedDict
 
 import cherrypy
 import humanfriendly
 import psutil
-from wtforms import validators, widgets
-from wtforms.fields.core import StringField, SelectField, Field
-from wtforms.fields.html5 import EmailField
-from wtforms.fields.simple import PasswordField
-
 from rdiffweb.controller import Controller, flash
 from rdiffweb.controller.cherrypy_wtf import CherryForm
 from rdiffweb.core.config import Option
 from rdiffweb.core.i18n import ugettext as _
 from rdiffweb.core.quota import QuotaUnsupported
 from rdiffweb.core.store import ADMIN_ROLE, MAINTAINER_ROLE, USER_ROLE
-from collections import OrderedDict
+from wtforms import validators, widgets
+from wtforms.fields.core import Field, SelectField, StringField
+from wtforms.fields.html5 import EmailField
+from wtforms.fields.simple import PasswordField
 
 # Define the logger
 logger = logging.getLogger(__name__)
@@ -49,12 +48,18 @@ def get_pyinfo():
         yield _('OS Version'), '%s %s (%s %s)' % (platform.system(), platform.release(), distro.linux_distribution()[0].capitalize(), distro.linux_distribution()[1])
     except:
         yield _('OS Version'), '%s %s' % (platform.system(), platform.release())
-    if hasattr(os, 'path'): yield _('OS Path'), os.environ['PATH']
-    if hasattr(sys, 'version'): yield _('Python Version'), ''.join(sys.version)
-    if hasattr(sys, 'subversion'): yield _('Python Subversion'), ', '.join(sys.subversion)
-    if hasattr(sys, 'prefix'): yield _('Python Prefix'), sys.prefix
-    if hasattr(sys, 'executable'): yield _('Python Executable'), sys.executable
-    if hasattr(sys, 'path'): yield _('Python Path'), ', '.join(sys.path)
+    if hasattr(os, 'path'):
+        yield _('OS Path'), os.environ['PATH']
+    if hasattr(sys, 'version'):
+        yield _('Python Version'), ''.join(sys.version)
+    if hasattr(sys, 'subversion'):
+        yield _('Python Subversion'), ', '.join(sys.subversion)
+    if hasattr(sys, 'prefix'):
+        yield _('Python Prefix'), sys.prefix
+    if hasattr(sys, 'executable'):
+        yield _('Python Executable'), sys.executable
+    if hasattr(sys, 'path'):
+        yield _('Python Path'), ', '.join(sys.path)
 
 
 def get_osinfo():
@@ -71,7 +76,8 @@ def pw_name(uid):
         except:
             return
 
-    if hasattr(sys, 'getfilesystemencoding'): yield _('File System Encoding'), sys.getfilesystemencoding()
+    if hasattr(sys, 'getfilesystemencoding'):
+        yield _('File System Encoding'), sys.getfilesystemencoding()
     if hasattr(os, 'getcwd'):
         yield _('Current Working Directory'), os.getcwd()
     if hasattr(os, 'getegid'):
@@ -167,10 +173,9 @@ class UserForm(CherryForm):
         validators=[validators.optional()],
         description=_("Disk spaces (in bytes) used by this user."))
 
-    @property
-    def error_message(self):
-        if self.errors:
-            return ' '.join(['%s: %s' % (field, ', '.join(messages)) for field, messages in self.errors.items()])
+
+class DeleteUserForm(CherryForm):
+    username = StringField(_('Username'), validators=[validators.required()])
 
 
 @cherrypy.tools.is_admin()
@@ -180,6 +185,67 @@ class AdminPage(Controller):
     logfile = Option('log_file')
     logaccessfile = Option('log_access_file')
 
+    def _add_or_edit_user(self, action, form):
+        assert action in ['add', 'edit']
+        assert form
+        # Validate form.
+        if not form.validate():
+            flash(form.error_message, level='error')
+            return
+        if action == 'add':
+            user = self.app.store.add_user(form.username.data, form.password.data)
+        else:
+            user = self.app.store.get_user(form.username.data)
+            if form.password.data:
+                user.set_password(form.password.data, old_password=None)
+        # Don't allow the user to changes it's "role" state.
+        if form.username.data != self.app.currentuser.username:
+            user.role = form.role.data
+        user.email = form.email.data or ''
+        if form.user_root.data:
+            user.user_root = form.user_root.data
+            if not user.valid_user_root():
+                flash(_("User's root directory %s is not accessible!") % user.user_root, level='error')
+                logger.warning("user's root directory %s is not accessible" % user.user_root)
+        # Try to update disk quota if the human readable value changed.
+        # Report error using flash.
+        if form.disk_quota and form.disk_quota.data:
+            try:
+                new_quota = form.disk_quota.data
+                old_quota = humanfriendly.parse_size(humanfriendly.format_size(user.disk_quota, binary=True))
+                if old_quota != new_quota:
+                    user.disk_quota = new_quota
+                    flash(_("User's quota updated"), level='success')
+            except QuotaUnsupported as e:
+                flash(_("Setting user's quota is not supported"), level='warning')
+            except Exception as e:
+                flash(_("Failed to update user's quota: %s") % e, level='error')
+                logger.warning("failed to update user's quota", exc_info=1)
+        if action == 'add':
+            flash(_("User added successfully."))
+        else:
+            flash(_("User information modified successfully."))
+
+    def _delete_user(self, action, form):
+        assert action == 'delete'
+        assert form
+        # Validate form.
+        if not form.validate():
+            flash(form.error_message, level='error')
+            return
+        if form.username.data == self.app.currentuser.username:
+            flash(_("You cannot remove your own account!"), level='error')
+        else:
+            try:
+                user = self.app.store.get_user(form.username.data)
+                if user:
+                    user.delete()
+                    flash(_("User account removed."))
+                else:
+                    flash(_("User doesn't exists!"), level='warning')
+            except ValueError as e:
+                flash(e, level='error')
+
     def _get_log_files(self):
         """
         Return a list of log files to be shown in admin area.
@@ -226,60 +292,13 @@ def users(self, criteria=u"", search=u"", action=u"", **kwargs):
 
         # If we're just showing the initial page, just do that
         form = UserForm()
-        if action in ["add", "edit"] and not form.validate():
-            # Display the form error to user using flash
-            flash(form.error_message, level='error')
-
-        elif action in ["add", "edit"] and form.validate():
-            if action == 'add':
-                user = self.app.store.add_user(form.username.data, form.password.data)
-            else:
-                user = self.app.store.get_user(form.username.data)
-                if form.password.data:
-                    user.set_password(form.password.data, old_password=None)
-            # Don't allow the user to changes it's "role" state.
-            if form.username.data != self.app.currentuser.username:
-                user.role = form.role.data
-            user.email = form.email.data or ''
-            if form.user_root.data:
-                user.user_root = form.user_root.data
-                if not user.valid_user_root():
-                    flash(_("User's root directory %s is not accessible!") % user.user_root, level='error')
-                    logger.warning("user's root directory %s is not accessible" % user.user_root)
-            # Try to update disk quota if the human readable value changed.
-            # Report error using flash.
-            if form.disk_quota and form.disk_quota.data:
-                try:
-                    new_quota = form.disk_quota.data
-                    old_quota = humanfriendly.parse_size(humanfriendly.format_size(user.disk_quota, binary=True))
-                    if old_quota != new_quota:
-                        user.disk_quota = new_quota
-                        flash(_("User's quota updated"), level='success')
-                except QuotaUnsupported as e:
-                    flash(_("Setting user's quota is not supported"), level='warning')
-                except Exception as e:
-                    flash(_("Failed to update user's quota: %s") % e, level='error')
-                    logger.warning("failed to update user's quota", exc_info=1)
-            if action == 'add':
-                flash(_("User added successfully."))
-            else:
-                flash(_("User information modified successfully."))
+        if action in ["add", "edit"]:
+            self._add_or_edit_user(action, form)
         elif action == 'delete':
-            if form.username.data == self.app.currentuser.username:
-                flash(_("You cannot remove your own account!"), level='error')
-            else:
-                try:
-                    user = self.app.store.get_user(form.username.data)
-                    if user:
-                        user.delete()
-                        flash(_("User account removed."))
-                    else:
-                        flash(_("User doesn't exists!"), level='warning')
-                except ValueError as e:
-                    flash(e, level='error')
+            self._delete_user(action, DeleteUserForm())
 
         params = {
-            "form": UserForm(formdata=None),
+            "form": form,
             "criteria": criteria,
             "search": search,
             "users": list(self.app.store.users(search=search, criteria=criteria))}

rdiffweb/controller/pref_sshkeys.py

@@ -47,7 +47,7 @@ def validate_key(unused_form, field):
         raise ValidationError(_("Invalid SSH key."))
 
 
-class SSHForm(CherryForm):
+class SshForm(CherryForm):
     title = StringField(
         _('Title'),
         description=_('The title is an optional description to identify the key. e.g.: bob@thinkpad-t530'),
@@ -60,6 +60,10 @@ class SSHForm(CherryForm):
     fingerprint = StringField('Fingerprint')
 
 
+class DeleteSshForm(CherryForm):
+    fingerprint = StringField('Fingerprint')
+
+
 class SSHKeysPlugin(Controller):
     """
     Plugin to configure SSH keys.
@@ -69,30 +73,45 @@ class SSHKeysPlugin(Controller):
 
     panel_name = _('SSH Keys')
 
+    def _add_key(self, action, form):
+        assert action == 'add'
+        assert form
+        if not form.validate():
+            for unused, messages in form.errors.items():
+                for message in messages:
+                    flash(message, level='warning')
+            return
+        try:
+            self.app.currentuser.add_authorizedkey(key=form.key.data, comment=form.title.data)
+        except DuplicateSSHKeyError as e:
+            flash(str(e), level='error')
+        except:
+            flash(_("Unknown error while adding the SSH Key"), level='error')
+            _logger.warning("error adding ssh key", exc_info=1)
+
+    def _delete_key(self, action, form):
+        assert action == 'delete'
+        assert form
+        if not form.validate():
+            for unused, messages in form.errors.items():
+                for message in messages:
+                    flash(message, level='warning')
+            return
+        is_maintainer()
+        try:
+            self.app.currentuser.delete_authorizedkey(form.fingerprint.data)
+        except:
+            flash(_("Unknown error while removing the SSH Key"), level='error')
+            _logger.warning("error removing ssh key", exc_info=1)
+
     def render_prefs_panel(self, panelid, action=None, **kwargs):  # @UnusedVariable
 
         # Handle action
-        form = SSHForm()
-        if action == "add" and not form.validate():
-            for unused_field, messages in form.errors.items():
-                for message in messages:
-                    flash(message, level='warning')
-        elif action == 'add':
-            # Add the key to the current user.
-            try:
-                self.app.currentuser.add_authorizedkey(key=form.key.data, comment=form.title.data)
-            except DuplicateSSHKeyError as e:
-                flash(str(e), level='error')
-            except:
-                flash(_("Unknown error while adding the SSH Key"), level='error')
-                _logger.warning("error adding ssh key", exc_info=1)
+        form = SshForm()
+        if action == "add":
+            self._add_key(action, form)
         elif action == 'delete':
-            is_maintainer()
-            try:
-                self.app.currentuser.delete_authorizedkey(form.fingerprint.data)
-            except:
-                flash(_("Unknown error while removing the SSH Key"), level='error')
-                _logger.warning("error removing ssh key", exc_info=1)
+            self._delete_key(action, DeleteSshForm())
 
         # Get SSH keys if file exists.
         params = {