CVE-2022-3456: Allocation of Resources Without Limits or Throttling (CWE-770)
value: Allocation of Resources Without Limits or Throttling in GitHub repository ikus060/rdiffweb prior to 2.5.0.
CWE description: The product allocates a reusable resource or group of resources on behalf of an actor without imposing any restrictions on the size or number of resources that can be allocated, in violation of the intended security policy for that actor.
CVSS v3 base score: 9.8
Published: 2022-10-13T20:15Z
Repository: https://github.com/ikus060/rdiffweb
Language: Python
Files changed: README.md, doc/configuration.md, rdiffweb/controller/api.py, rdiffweb/controller/form.py, rdiffweb/controller/page_admin_session.py, rdiffweb/controller/page_login.py, rdiffweb/controller/page_mfa.py, rdiffweb/controller/page_pref_general.py, rdiffweb/controller/page_pref_session.py, rdiffweb/controller/tests/test_api.py, rdiffweb/controller/tests/test_page_login.py, rdiffweb/controller/tests/test_page_prefs_general.py, rdiffweb/core/config.py, rdiffweb/rdw_app.py, rdiffweb/tools/ratelimit.py
Commit date: 2022-10-11 14:46:08 -0400
Commit message: Improve ratelimit implementation

Diff: diff --git a/README.md b/README.md
index 09c51a982..68558217d 100644
--- a/README.md
+++ b/README.md
@@ -132,10 +132,11 @@ This next release focus on two-factor-authentication as a measure to increase se
 * Generate a new session on login and 2FA #220
 * Enforce permission on /etc/rdiffweb configuration folder
 * Enforce validation on fullname, username and email
 * Limit incorrect attempts to change the user's password to prevent brute force attacks #225 [CVE-2022-3273](https://nvd.nist.gov/vuln/detail/CVE-2022-3273)
 * Enforce password policy new password cannot be set as new password [CVE-2022-3376](https://nvd.nist.gov/vuln/detail/CVE-2022-3376)
+* 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)
 
 Breaking changes:
 
 * Drop Ubuntu Hirsute & Impish (End-of-life)
 * `session-dir` is deprecated and should be replace by `rate-limit-dir`. User's session are stored in database.
diff --git a/doc/configuration.md b/doc/configuration.md
index 680723d29..ec795b059 100644
--- a/doc/configuration.md
+++ b/doc/configuration.md
@@ -271,11 +271,11 @@ Take note, it's better to enable project quota attributes when the repositories
 Rdiffweb could be configured to rate-limit access to anonymous to avoid bruteforce
 attacks and authenticated users to avoid Denial Of Service attack.
 
 | Option | Description | Example |
 | --- | --- | --- |
-| rate-limit | maximum number of requests per minute that can be made by an IP address for an unauthenticated connection. When this limit is reached, an HTTP 429 message is returned to the user. This security measure is used to limit brute force attacks on the login page and the RESTful API. | 10 |
+| rate-limit | maximum number of requests per hour that can be made on sensitive endpoints. When this limit is reached, an HTTP 429 message is returned to the user or the user is logged out. This security measure is used to limit brute force attacks on the login page and the RESTful API. | 20 |
 | rate-limit-dir | location where to store rate-limit information. When undefined, data is kept in memory. | /var/lib/rdiffweb/session |
 
 ## Custom user's password length limits
 
 By default, Rdiffweb supports passwords with the following lengths:
diff --git a/rdiffweb/controller/api.py b/rdiffweb/controller/api.py
index 04c20acb2..9b1e2bf6f 100644
--- a/rdiffweb/controller/api.py
+++ b/rdiffweb/controller/api.py
@@ -62,13 +62,18 @@ def _checkpassword(realm, username, password):
         # Verify if the password matches a token.
         if userobj.validate_access_token(password):
             return True
         # Disable password authentication for MFA
         if userobj.mfa == UserObject.ENABLED_MFA:
+            cherrypy.tools.ratelimit.hit()
             return False
     # Otherwise validate username password
-    return any(cherrypy.engine.publish('login', username, password))
+    valid = any(cherrypy.engine.publish('login', username, password))
+    if not valid:
+        # When invalid, we need to increase the rate limit.
+        cherrypy.tools.ratelimit.hit()
+    return valid
 
 
 class ApiCurrentUser(Controller):
     @cherrypy.expose
     def default(self):
@@ -97,13 +102,13 @@ class ApiCurrentUser(Controller):
 @cherrypy.tools.json_out(handler=json_handler)
 @cherrypy.config(**{'error_page.default': False})
 @cherrypy.tools.auth_basic(realm='rdiffweb', checkpassword=_checkpassword, priority=70)
 @cherrypy.tools.auth_form(on=False)
 @cherrypy.tools.auth_mfa(on=False)
-@cherrypy.tools.sessions(on=False)
 @cherrypy.tools.i18n(on=False)
-@cherrypy.tools.ratelimit()
+@cherrypy.tools.ratelimit(scope='rdiffweb-api', hit=0, priority=69)
+@cherrypy.tools.sessions(on=False)
 class ApiPage(Controller):
     """
     This class provide a restful API to access some of the rdiffweb resources.
     """
 
diff --git a/rdiffweb/controller/form.py b/rdiffweb/controller/form.py
index 1770da180..f5c4562e5 100644
--- a/rdiffweb/controller/form.py
+++ b/rdiffweb/controller/form.py
@@ -17,12 +17,10 @@
 
 import cherrypy
 from markupsafe import Markup
 from wtforms.form import Form
 
-SUBMIT_METHODS = {'POST', 'PUT', 'PATCH', 'DELETE'}
-
 
 class _ProxyFormdata:
     """
     Custom class to proxy default form data into WTForm from cherrypy variables.
     """
@@ -63,11 +61,11 @@ class CherryForm(Form):
     def is_submitted(self):
         """
         Consider the form submitted if there is an active request and
         the method is ``POST``, ``PUT``, ``PATCH``, or ``DELETE``.
         """
-        return cherrypy.request.method in SUBMIT_METHODS
+        return cherrypy.request.method == 'POST'
 
     def validate_on_submit(self):
         """
         Call `validate` only if the form is submitted.
         This is a shortcut for ``form.is_submitted() and form.validate()``.
diff --git a/rdiffweb/controller/page_admin_session.py b/rdiffweb/controller/page_admin_session.py
index 2f56d559e..065167ddf 100644
--- a/rdiffweb/controller/page_admin_session.py
+++ b/rdiffweb/controller/page_admin_session.py
@@ -28,14 +28,10 @@ from rdiffweb.tools.i18n import ugettext as _
 
 class RevokeSessionForm(CherryForm):
     action = StringField(validators=[validators.regexp('delete')])
     number = IntegerField(validators=[validators.data_required()])
 
-    @property
-    def app(self):
-        return cherrypy.request.app
-
 
 @cherrypy.tools.is_admin()
 class AdminSessionPage(Controller):
     @cherrypy.expose
     def default(self, action=None, **kwargs):
diff --git a/rdiffweb/controller/page_login.py b/rdiffweb/controller/page_login.py
index ed3daa3b8..c20038118 100644
--- a/rdiffweb/controller/page_login.py
+++ b/rdiffweb/controller/page_login.py
@@ -60,11 +60,11 @@ class LoginPage(Controller):
     This page is used by the authentication to enter a user/pass.
     """
 
     @cherrypy.expose()
     @cherrypy.tools.auth_mfa(on=False)
-    @cherrypy.tools.ratelimit()
+    @cherrypy.tools.ratelimit(methods=['POST'])
     def index(self, **kwargs):
         """
         Called by auth_form to generate the /login/ page.
         """
         form = LoginForm()
diff --git a/rdiffweb/controller/page_mfa.py b/rdiffweb/controller/page_mfa.py
index 3f8ec37c8..9b6981d21 100644
--- a/rdiffweb/controller/page_mfa.py
+++ b/rdiffweb/controller/page_mfa.py
@@ -68,11 +68,11 @@ class MfaForm(CherryForm):
         return super().validate()
 
 
 class MfaPage(Controller):
     @cherrypy.expose()
-    @cherrypy.tools.ratelimit()
+    @cherrypy.tools.ratelimit(methods=['POST'])
     def index(self, **kwargs):
         form = MfaForm()
 
         # Validate MFA
         if form.is_submitted():
diff --git a/rdiffweb/controller/page_pref_general.py b/rdiffweb/controller/page_pref_general.py
index 9ae459e2f..eb46eda68 100644
--- a/rdiffweb/controller/page_pref_general.py
+++ b/rdiffweb/controller/page_pref_general.py
@@ -27,14 +27,10 @@ from wtforms.validators import DataRequired, EqualTo, InputRequired, Length, Opt
 from rdiffweb.controller import Controller, flash
 from rdiffweb.controller.form import CherryForm
 from rdiffweb.core.model import UserObject
 from rdiffweb.tools.i18n import gettext_lazy as _
 
-# Maximum number of password change attempt before logout
-CHANGE_PASSWORD_MAX_ATTEMPT = 5
-CHANGE_PASSWORD_ATTEMPTS = 'change_password_attempts'
-
 
 class UserProfileForm(CherryForm):
     action = HiddenField(default='set_profile_info')
     username = StringField(_('Username'), render_kw={'readonly': True})
     fullname = StringField(
@@ -97,27 +93,12 @@ class UserPasswordForm(CherryForm):
 
     def populate_obj(self, user):
         # Check if current password is "valid" if Not, rate limit the
         # number of attempts and logout user after too many invalid attempts.
         if not user.validate_password(self.current.data):
-            cherrypy.session[CHANGE_PASSWORD_ATTEMPTS] = cherrypy.session.get(CHANGE_PASSWORD_ATTEMPTS, 0) + 1
-            attempts = cherrypy.session[CHANGE_PASSWORD_ATTEMPTS]
-            if attempts >= CHANGE_PASSWORD_MAX_ATTEMPT:
-                cherrypy.session.clear()
-                cherrypy.session.regenerate()
-                flash(
-                    _("You were logged out because you entered the wrong password too many times."),
-                    level='warning',
-                )
-                raise cherrypy.HTTPRedirect('/login/')
             self.current.errors = [_("Wrong current password.")]
             return False
-
-        # Clear number of attempts
-        if CHANGE_PASSWORD_ATTEMPTS in cherrypy.session:
-            del cherrypy.session[CHANGE_PASSWORD_ATTEMPTS]
-
         try:
             user.set_password(self.new.data)
             return True
         except ValueError as e:
             self.new.errors = [str(e)]
@@ -149,10 +130,11 @@ class PagePrefsGeneral(Controller):
     """
     Plugin to change user profile and password.
     """
 
     @cherrypy.expose
+    @cherrypy.tools.ratelimit(methods=['POST'], logout=True)
     def default(self, **kwargs):
         # Process the parameters.
         profile_form = UserProfileForm(obj=self.app.currentuser)
         password_form = UserPasswordForm()
         refresh_form = RefreshForm()
diff --git a/rdiffweb/controller/page_pref_session.py b/rdiffweb/controller/page_pref_session.py
index 7a72df3cd..72094e26c 100644
--- a/rdiffweb/controller/page_pref_session.py
+++ b/rdiffweb/controller/page_pref_session.py
@@ -28,14 +28,10 @@ from rdiffweb.tools.i18n import ugettext as _
 
 class RevokeSessionForm(CherryForm):
     action = StringField(validators=[validators.regexp('delete')])
     number = IntegerField(validators=[validators.data_required()])
 
-    @property
-    def app(self):
-        return cherrypy.request.app
-
 
 class PagePrefSession(Controller):
     @cherrypy.expose
     def default(self, action=None, **kwargs):
         # Delete session on form submit
diff --git a/rdiffweb/controller/tests/test_api.py b/rdiffweb/controller/tests/test_api.py
index 163ed949b..d7b0a846a 100644
--- a/rdiffweb/controller/tests/test_api.py
+++ b/rdiffweb/controller/tests/test_api.py
@@ -122,8 +122,16 @@ class APIRatelimitTest(rdiffweb.test.WebCase):
     default_config = {
         'rate-limit': 5,
     }
 
     def test_login_ratelimit(self):
-        for i in range(0, 6):
-            self.getPage('/api/')
+        # Given invalid credentials sent to API
+        headers = [("Authorization", "Basic " + b64encode(b"admin:invalid").decode('ascii'))]
+        for i in range(1, 5):
+            self.getPage('/api/', headers=headers)
+            self.assertStatus(401)
+        # Then the 6th request is refused
+        self.getPage('/api/', headers=headers)
+        self.assertStatus(429)
+        # Next request is also refused event if credentials are valid.
+        self.getPage('/api/', headers=[("Authorization", "Basic " + b64encode(b"admin:admin123").decode('ascii'))])
         self.assertStatus(429)
diff --git a/rdiffweb/controller/tests/test_page_login.py b/rdiffweb/controller/tests/test_page_login.py
index 9de264b9d..9f2b760d6 100644
--- a/rdiffweb/controller/tests/test_page_login.py
+++ b/rdiffweb/controller/tests/test_page_login.py
@@ -17,13 +17,13 @@
 """
 Created on Dec 26, 2015
 
 @author: Patrik Dufresne
 """
+import os
 
-
-from parameterized import parameterized
+from parameterized import parameterized, parameterized_class
 
 import rdiffweb.test
 from rdiffweb.core.model import DbSession, SessionObject, UserObject
 from rdiffweb.tools.auth_form import LOGIN_TIME, SESSION_KEY
 
@@ -210,71 +210,84 @@ class LoginPageWithHeaderName(rdiffweb.test.WebCase):
         # Then the page display the header-name
         self.assertStatus('200 OK')
         self.assertInBody('HEADER-NAME')
 
 
+@parameterized_class(
+    [
+        {"default_config": {'rate-limit': 5}},
+        {"default_config": {'rate-limit': 5, 'rate-limit-dir': '/tmp'}},
+    ]
+)
 class LoginPageRateLimitTest(rdiffweb.test.WebCase):
-
-    default_config = {
-        'rate-limit': 5,
-    }
+    def setUp(self):
+        if os.path.isfile('/tmp/ratelimit-127.0.0.1'):
+            os.unlink('/tmp/ratelimit-127.0.0.1')
+        if os.path.isfile('/tmp/ratelimit-127.0.0.1.-login'):
+            os.unlink('/tmp/ratelimit-127.0.0.1.-login')
+        return super().setUp()
 
     def test_login_ratelimit(self):
         # Given an unauthenticate
-        # When requesting multple time the login page
-        for i in range(0, 6):
-            self.getPage('/login/')
+        # When requesting multiple time the login page
+        for i in range(1, 5):
+            self.getPage('/login/', method='POST', body={'login': 'invalid', 'password': 'invalid'})
+            self.assertStatus(200)
         # Then a 429 error (too many request) is return
+        self.getPage('/login/', method='POST', body={'login': 'invalid', 'password': 'invalid'})
         self.assertStatus(429)
 
 
-class LoginPageRateLimitWithSessionDirTest(rdiffweb.test.WebCase):
+class LoginPageRateLimitTest2(rdiffweb.test.WebCase):
 
-    default_config = {
-        'rate-limit-dir': '/tmp',
-        'rate-limit': 5,
-    }
+    default_config = {'rate-limit': 5}
 
-    def test_login_ratelimit(self):
+    def test_login_ratelimit_forwarded_for(self):
         # Given an unauthenticate
-        # When requesting multple time the login page
-        for i in range(0, 6):
-            self.getPage('/login/')
-        # Then a 429 error (too many request) is return
+        # When requesting multiple time the login page with different `X-Forwarded-For`
+        for i in range(1, 5):
+            self.getPage(
+                '/login/',
+                headers=[('X-Forwarded-For', '127.0.0.%s' % i)],
+                method='POST',
+                body={'login': 'invalid', 'password': 'invalid'},
+            )
+            self.assertStatus(200)
+        # Then original IP get blocked
+        self.getPage(
+            '/login/',
+            headers=[('X-Forwarded-For', '127.0.0.%s' % i)],
+            method='POST',
+            body={'login': 'invalid', 'password': 'invalid'},
+        )
         self.assertStatus(429)
 
 
-class LoginPageRateLimitTestWithXForwardedFor(rdiffweb.test.WebCase):
+class LoginPageRateLimitTest3(rdiffweb.test.WebCase):
+    default_config = {'rate-limit': 5}
 
-    default_config = {
-        'rate-limit': 5,
-    }
-
-    def test_login_ratelimit(self):
+    def test_login_ratelimit_real_ip(self):
         # Given an unauthenticate
-        # When requesting multple time the login page
-        for i in range(0, 6):
-            self.getPage('/login/', headers=[('X-Forwarded-For', '127.0.0.%s' % i)])
-        # Then a 429 error (too many request) is return
+        # When requesting multiple time the login page with different `X-Real-IP`
+        for i in range(1, 5):
+            self.getPage(
+                '/login/',
+                headers=[('X-Real-IP', '127.0.0.128')],
+                method='POST',
+                body={'login': 'invalid', 'password': 'invalid'},
+            )
+            self.assertStatus(200)
+        # Then the X-Real-IP get blocked
+        self.getPage(
+            '/login/',
+            headers=[('X-Real-IP', '127.0.0.128')],
+            method='POST',
+            body={'login': 'invalid', 'password': 'invalid'},
+        )
         self.assertStatus(429)
 
 
-class LoginPageRateLimitTestWithXRealIP(rdiffweb.test.WebCase):
-
-    default_config = {
-        'rate-limit': 5,
-    }
-
-    def test_login_ratelimit(self):
-        # Given an unauthenticate
-        # When requesting multple time the login page
-        for i in range(0, 6):
-            self.getPage('/login/', headers=[('X-Real-IP', '127.0.0.%s' % i)])
-        # Then a 200 is return.
-        self.assertStatus(200)
-
-
 class LogoutPageTest(rdiffweb.test.WebCase):
     def test_getpage_without_login(self):
         # Given an unauthenticated user
         # When Accessing logout page directly
         self.getPage('/logout')
diff --git a/rdiffweb/controller/tests/test_page_prefs_general.py b/rdiffweb/controller/tests/test_page_prefs_general.py
index be60757cf..940f60504 100644
--- a/rdiffweb/controller/tests/test_page_prefs_general.py
+++ b/rdiffweb/controller/tests/test_page_prefs_general.py
@@ -201,25 +201,10 @@ class PagePrefGeneralTest(rdiffweb.test.WebCase):
     def test_change_password_with_too_long(self):
         new_password = 'a' * 129
         self._set_password(self.PASSWORD, new_password, new_password)
         self.assertInBody("Password must have between 8 and 128 characters.")
 
-    def test_change_password_too_many_attemps(self):
-        # When udating user's password with wrong current password 5 times
-        for _i in range(1, 5):
-            self._set_password('wrong', "pr3j5Dwi", "pr3j5Dwi")
-            self.assertStatus(200)
-            self.assertInBody("Wrong current password.")
-        # Then user session is cleared and user is redirect to login page
-        self._set_password('wrong', "pr3j5Dwi", "pr3j5Dwi")
-        self.assertStatus(303)
-        self.assertHeaderItemValue('Location', self.baseurl + '/login/')
-        # Then a warning message is displayed on login page
-        self.getPage('/login/')
-        self.assertStatus(200)
-        self.assertInBody('You were logged out because you entered the wrong password too many times.')
-
     def test_change_password_with_same_value(self):
         # Given a user with a password
         self._set_password(self.PASSWORD, "pr3j5Dwi", "pr3j5Dwi")
         self.assertStatus(303)
         # When updating the pasword with the same password
@@ -254,5 +239,33 @@ class PagePrefGeneralTest(rdiffweb.test.WebCase):
         # Then a success message is displayed
         self.assertInBody('Repositories successfully updated')
         # Then the list is free of inexisting repos.
         userobj.expire()
         self.assertEqual(['broker-repo', 'testcases'], sorted([r.name for r in userobj.repo_objs]))
+
+
+class PagePrefGeneralRateLimitTest(rdiffweb.test.WebCase):
+    login = True
+
+    default_config = {'rate-limit': 5}
+
+    def test_change_password_too_many_attemps(self):
+        # When udating user's password with wrong current password 5 times
+        for _i in range(1, 5):
+            self.getPage(
+                '/prefs/general',
+                method='POST',
+                body={'action': 'set_password', 'current': 'wrong', 'new': 'pr3j5Dwi', 'confirm': 'pr3j5Dwi'},
+            )
+            self.assertStatus(200)
+            self.assertInBody("Wrong current password.")
+        # Then user session is cleared and user is redirect to login page
+        self.getPage(
+            '/prefs/general',
+            method='POST',
+            body={'action': 'set_password', 'current': 'wrong', 'new': 'pr3j5Dwi', 'confirm': 'pr3j5Dwi'},
+        )
+        self.assertStatus(303)
+        self.assertHeaderItemValue('Location', self.baseurl + '/')
+        # Then a warning message is displayed on login page
+        self.getPage('/login/')
+        self.assertStatus(200)
diff --git a/rdiffweb/core/config.py b/rdiffweb/core/config.py
index f2fd1c297..b5eef33f8 100644
--- a/rdiffweb/core/config.py
+++ b/rdiffweb/core/config.py
@@ -402,12 +402,12 @@ def get_parser():
 
     parser.add(
         '--rate-limit',
         metavar='LIMIT',
         type=int,
-        default=30,
-        help='maximum number of requests per minute that can be made by an IP address for an unauthenticated connection. When this limit is reached, an HTTP 429 message is returned to the user. This security measure is used to limit brute force attacks on the login page and the RESTful API.',
+        default=20,
+        help='maximum number of requests per hour that can be made on sensitive endpoints. When this limit is reached, an HTTP 429 message is returned to the user or the user is logged out. This security measure is used to limit brute force attacks on the login page and the RESTful API.',
     )
 
     parser.add(
         '--ssl-certificate',
         '--sslcertificate',
diff --git a/rdiffweb/rdw_app.py b/rdiffweb/rdw_app.py
index 747891df5..edf022093 100644
--- a/rdiffweb/rdw_app.py
+++ b/rdiffweb/rdw_app.py
@@ -207,12 +207,12 @@ class RdiffwebApp(Application):
                 'tools.sessions.httponly': True,
                 'tools.sessions.timeout': cfg.session_timeout,  # minutes
                 'tools.sessions.persistent': False,  # auth_form should update this.
                 'tools.auth_form.timeout': cfg.session_persistent_timeout,  # minutes
                 'tools.ratelimit.debug': cfg.debug,
-                'tools.ratelimit.delay': 60,
-                'tools.ratelimit.anonymous_limit': cfg.rate_limit,
+                'tools.ratelimit.delay': 3600,
+                'tools.ratelimit.limit': cfg.rate_limit,
                 'tools.ratelimit.storage_class': rate_limit_storage_class,
                 'tools.ratelimit.storage_path': cfg.rate_limit_dir,
             },
         }
 
diff --git a/rdiffweb/tools/ratelimit.py b/rdiffweb/tools/ratelimit.py
index 9c5357134..ae39cd294 100644
--- a/rdiffweb/tools/ratelimit.py
+++ b/rdiffweb/tools/ratelimit.py
@@ -1,8 +1,8 @@
 # -*- coding: utf-8 -*-
-# rdiffweb, A web interface to rdiff-backup repositories
-# Copyright (C) 2012-2021 rdiffweb contributors
+# udb, A web interface to manage IT network
+# Copyright (C) 2022 IKUS Software inc.
 #
 # 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.
@@ -31,17 +31,17 @@ class _DataStore:
     """
 
     def __init__(self, **kwargs):
         self._locks = {}
 
-    def get_and_increment(self, token, delay):
+    def get_and_increment(self, token, delay, hit=1):
         lock = self._locks.setdefault(token, threading.RLock())
         with lock:
             tracker = self._load(token)
             if tracker is None or tracker.timeout < time.time():
                 tracker = Tracker(token=token, hits=0, timeout=int(time.time() + delay))
-            tracker = tracker._replace(hits=tracker.hits + 1)
+            tracker = tracker._replace(hits=tracker.hits + hit)
             self._save(tracker)
         return tracker.hits
 
     def _save(self, tracker):
         raise NotImplementedError
@@ -95,11 +95,11 @@ class FileRateLimit(_DataStore):
             f = open(path, 'rb')
             try:
                 return pickle.load(f)
             finally:
                 f.close()
-        except (IOError, EOFError):
+        except Exception:
             # Drop session data if invalid
             pass
         return None
 
     def _save(self, tracker):
@@ -109,45 +109,76 @@ class FileRateLimit(_DataStore):
             pickle.dump(tracker, f, self.pickle_protocol)
         finally:
             f.close()
 
 
-def check_ratelimit(delay=60, anonymous_limit=0, registered_limit=0, rate_exceed_status=429, debug=False, **conf):
+def check_ratelimit(
+    delay=3600, limit=25, return_status=429, logout=False, scope=None, methods=None, debug=False, hit=1, **conf
+):
     """
-    Verify the ratelimit. By default return a 429 HTTP error code (Too Many Request).
+    Verify the ratelimit. By default return a 429 HTTP error code (Too Many Request). After 25 request within the same hour.
+
+    Arguments:
+        delay:         Time window for analysis in seconds. Default per hour (3600 seconds)
+        limit:         Number of request allowed for an entry point. Default 25
+        return_status: HTTP Error code to return.
+        logout:        True to logout user when limit is reached
+        scope:         if specify, define the scope of rate limit. Default to path_info.
+        methods:       if specify, only the methods in the list will be rate limited.
+    """
+    assert delay > 0, 'invalid delay'
 
-    Usage:
+    # Check if limit is enabled
+    if limit <= 0:
+        return
 
-    @cherrypy.tools.ratelimit(on=True, anonymous_limit=5, registered_limit=50, storage_class=FileRateLimit, storage_path='/tmp')
-    def index(self):
-        pass
-    """
+    # Check if this 'method' should be rate limited
+    request = cherrypy.request
+    if methods is not None and request.method not in methods:
+        if debug:
+            cherrypy.log(
+                'skip rate limit for HTTP method %s' % (request.method,),
+                'TOOLS.RATELIMIT',
+            )
+        return
 
     # If datastore is not pass as configuration, create it for the first time.
-    datastore = getattr(cherrypy, '_ratelimit_datastore', None)
+    datastore = getattr(cherrypy.request.app, '_ratelimit_datastore', None)
     if datastore is None:
         # Create storage using storage class
         storage_class = conf.get('storage_class', RamRateLimit)
         datastore = storage_class(**conf)
-        cherrypy._ratelimit_datastore = datastore
+        cherrypy.request.app._ratelimit_datastore = datastore
 
     # If user is authenticated, use the username else use the ip address
-    token = cherrypy.request.login or cherrypy.request.remote.ip
-
-    # Get the real limit depending of user login.
-    limit = registered_limit if cherrypy.request.login else anonymous_limit
-    if limit is None or limit <= 0:
-        return
+    token = (request.login or request.remote.ip) + '.' + (scope or request.path_info)
 
     # Get hits count using datastore.
-    hits = datastore.get_and_increment(token, delay)
+    hits = datastore.get_and_increment(token, delay, hit)
     if debug:
         cherrypy.log(
-            'check and increase rate limit for token %s, limit %s, hits %s' % (token, limit, hits), 'TOOLS.RATELIMIT'
+            'check and increase rate limit for scope %s, limit %s, hits %s' % (token, limit, hits), 'TOOLS.RATELIMIT'
         )
 
     # Verify user has not exceeded rate limit
     if limit <= hits:
-        raise cherrypy.HTTPError(rate_exceed_status)
+        if logout:
+            if hasattr(cherrypy, 'session'):
+                cherrypy.session.clear()
+            raise cherrypy.HTTPRedirect("/")
+
+        raise cherrypy.HTTPError(return_status)
+
+
+def hit(hit=1):
+    """
+    May be called directly by handlers to add a hit for the given request.
+    """
+    conf = cherrypy.tools.ratelimit._merged_args()
+    conf['hit'] = hit
+    cherrypy.tools.ratelimit.callable(**conf)
 
 
 cherrypy.tools.ratelimit = cherrypy.Tool('before_handler', check_ratelimit, priority=60)
+
+
+cherrypy.tools.ratelimit.hit = hit

Vulnerable code: | rate-limit | maximum number of requests per minute that can be made by an IP address for an unauthenticated connection. When this limit is reached, an HTTP 429 message is returned to the user. This security measure is used to limit brute force attacks on the login page and the RESTful API. | 10 |
    return any(cherrypy.engine.publish('login', username, password))
@cherrypy.tools.sessions(on=False)
@cherrypy.tools.ratelimit()
SUBMIT_METHODS = {'POST', 'PUT', 'PATCH', 'DELETE'}

        return cherrypy.request.method in SUBMIT_METHODS
    @property
    def app(self):
        return cherrypy.request.app

    @cherrypy.tools.ratelimit()
    @cherrypy.tools.ratelimit()
# Maximum number of password change attempt before logout
CHANGE_PASSWORD_MAX_ATTEMPT = 5
CHANGE_PASSWORD_ATTEMPTS = 'change_password_attempts'

            cherrypy.session[CHANGE_PASSWORD_ATTEMPTS] = cherrypy.session.get(CHANGE_PASSWORD_ATTEMPTS, 0) + 1
            attempts = cherrypy.session[CHANGE_PASSWORD_ATTEMPTS]
            if attempts >= CHANGE_PASSWORD_MAX_ATTEMPT:
                cherrypy.session.clear()
                cherrypy.session.regenerate()
                flash(
                    _("You were logged out because you entered the wrong password too many times."),
                    level='warning',
                )
                raise cherrypy.HTTPRedirect('/login/')

        # Clear number of attempts
        if CHANGE_PASSWORD_ATTEMPTS in cherrypy.session:
            del cherrypy.session[CHANGE_PASSWORD_ATTEMPTS]

    @property
    def app(self):
        return cherrypy.request.app

        for i in range(0, 6):
            self.getPage('/api/')

from parameterized import parameterized

    default_config = {
        'rate-limit': 5,
    }
        # When requesting multple time the login page
        for i in range(0, 6):
            self.getPage('/login/')
class LoginPageRateLimitWithSessionDirTest(rdiffweb.test.WebCase):
    default_config = {
        'rate-limit-dir': '/tmp',
        'rate-limit': 5,
    }
    def test_login_ratelimit(self):
        # When requesting multple time the login page
        for i in range(0, 6):
            self.getPage('/login/')
        # Then a 429 error (too many request) is return
class LoginPageRateLimitTestWithXForwardedFor(rdiffweb.test.WebCase):
    default_config = {
        'rate-limit': 5,
    }

    def test_login_ratelimit(self):
        # When requesting multple time the login page
        for i in range(0, 6):
            self.getPage('/login/', headers=[('X-Forwarded-For', '127.0.0.%s' % i)])
        # Then a 429 error (too many request) is return
class LoginPageRateLimitTestWithXRealIP(rdiffweb.test.WebCase):

    default_config = {
        'rate-limit': 5,
    }

    def test_login_ratelimit(self):
        # Given an unauthenticate
        # When requesting multple time the login page
        for i in range(0, 6):
            self.getPage('/login/', headers=[('X-Real-IP', '127.0.0.%s' % i)])
        # Then a 200 is return.
        self.assertStatus(200)


    def test_change_password_too_many_attemps(self):
        # When udating user's password with wrong current password 5 times
        for _i in range(1, 5):
            self._set_password('wrong', "pr3j5Dwi", "pr3j5Dwi")
            self.assertStatus(200)
            self.assertInBody("Wrong current password.")
        # Then user session is cleared and user is redirect to login page
        self._set_password('wrong', "pr3j5Dwi", "pr3j5Dwi")
        self.assertStatus(303)
        self.assertHeaderItemValue('Location', self.baseurl + '/login/')
        # Then a warning message is displayed on login page
        self.getPage('/login/')
        self.assertStatus(200)
        self.assertInBody('You were logged out because you entered the wrong password too many times.')

        default=30,
        help='maximum number of requests per minute that can be made by an IP address for an unauthenticated connection. When this limit is reached, an HTTP 429 message is returned to the user. This security measure is used to limit brute force attacks on the login page and the RESTful API.',
                'tools.ratelimit.delay': 60,
                'tools.ratelimit.anonymous_limit': cfg.rate_limit,
# rdiffweb, A web interface to rdiff-backup repositories
# Copyright (C) 2012-2021 rdiffweb contributors
    def get_and_increment(self, token, delay):
            tracker = tracker._replace(hits=tracker.hits + 1)
        except (IOError, EOFError):
def check_ratelimit(delay=60, anonymous_limit=0, registered_limit=0, rate_exceed_status=429, debug=False, **conf):
    Verify the ratelimit. By default return a 429 HTTP error code (Too Many Request).
    Usage:
    @cherrypy.tools.ratelimit(on=True, anonymous_limit=5, registered_limit=50, storage_class=FileRateLimit, storage_path='/tmp')
    def index(self):
        pass
    """
    datastore = getattr(cherrypy, '_ratelimit_datastore', None)
        cherrypy._ratelimit_datastore = datastore
    token = cherrypy.request.login or cherrypy.request.remote.ip

    # Get the real limit depending of user login.
    limit = registered_limit if cherrypy.request.login else anonymous_limit
    if limit is None or limit <= 0:
        return
    hits = datastore.get_and_increment(token, delay)
            'check and increase rate limit for token %s, limit %s, hits %s' % (token, limit, hits), 'TOOLS.RATELIMIT'
        raise cherrypy.HTTPError(rate_exceed_status)
Fixed code: * 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)
| rate-limit | maximum number of requests per hour that can be made on sensitive endpoints. When this limit is reached, an HTTP 429 message is returned to the user or the user is logged out. This security measure is used to limit brute force attacks on the login page and the RESTful API. | 20 |
            cherrypy.tools.ratelimit.hit()
    valid = any(cherrypy.engine.publish('login', username, password))
    if not valid:
        # When invalid, we need to increase the rate limit.
        cherrypy.tools.ratelimit.hit()
    return valid
@cherrypy.tools.ratelimit(scope='rdiffweb-api', hit=0, priority=69)
@cherrypy.tools.sessions(on=False)
        return cherrypy.request.method == 'POST'
    @cherrypy.tools.ratelimit(methods=['POST'])
    @cherrypy.tools.ratelimit(methods=['POST'])
    @cherrypy.tools.ratelimit(methods=['POST'], logout=True)
        # Given invalid credentials sent to API
        headers = [("Authorization", "Basic " + b64encode(b"admin:invalid").decode('ascii'))]
        for i in range(1, 5):
            self.getPage('/api/', headers=headers)
            self.assertStatus(401)
        # Then the 6th request is refused
        self.getPage('/api/', headers=headers)
        self.assertStatus(429)
        # Next request is also refused event if credentials are valid.
        self.getPage('/api/', headers=[("Authorization", "Basic " + b64encode(b"admin:admin123").decode('ascii'))])
import os
from parameterized import parameterized, parameterized_class
@parameterized_class(
    [
        {"default_config": {'rate-limit': 5}},
        {"default_config": {'rate-limit': 5, 'rate-limit-dir': '/tmp'}},
    ]
)
    def setUp(self):
        if os.path.isfile('/tmp/ratelimit-127.0.0.1'):
            os.unlink('/tmp/ratelimit-127.0.0.1')
        if os.path.isfile('/tmp/ratelimit-127.0.0.1.-login'):
            os.unlink('/tmp/ratelimit-127.0.0.1.-login')
        return super().setUp()
        # When requesting multiple time the login page
        for i in range(1, 5):
            self.getPage('/login/', method='POST', body={'login': 'invalid', 'password': 'invalid'})
            self.assertStatus(200)
        self.getPage('/login/', method='POST', body={'login': 'invalid', 'password': 'invalid'})
class LoginPageRateLimitTest2(rdiffweb.test.WebCase):
    default_config = {'rate-limit': 5}
    def test_login_ratelimit_forwarded_for(self):
        # When requesting multiple time the login page with different `X-Forwarded-For`
        for i in range(1, 5):
            self.getPage(
                '/login/',
                headers=[('X-Forwarded-For', '127.0.0.%s' % i)],
                method='POST',
                body={'login': 'invalid', 'password': 'invalid'},
            )
            self.assertStatus(200)
        # Then original IP get blocked
        self.getPage(
            '/login/',
            headers=[('X-Forwarded-For', '127.0.0.%s' % i)],
            method='POST',
            body={'login': 'invalid', 'password': 'invalid'},
        )
class LoginPageRateLimitTest3(rdiffweb.test.WebCase):
    default_config = {'rate-limit': 5}
    def test_login_ratelimit_real_ip(self):
        # When requesting multiple time the login page with different `X-Real-IP`
        for i in range(1, 5):
            self.getPage(
                '/login/',
                headers=[('X-Real-IP', '127.0.0.128')],
                method='POST',
                body={'login': 'invalid', 'password': 'invalid'},
            )
            self.assertStatus(200)
        # Then the X-Real-IP get blocked
        self.getPage(
            '/login/',
            headers=[('X-Real-IP', '127.0.0.128')],
            method='POST',
            body={'login': 'invalid', 'password': 'invalid'},
        )


class PagePrefGeneralRateLimitTest(rdiffweb.test.WebCase):
    login = True

    default_config = {'rate-limit': 5}

    def test_change_password_too_many_attemps(self):
        # When udating user's password with wrong current password 5 times
        for _i in range(1, 5):
            self.getPage(
                '/prefs/general',
                method='POST',
                body={'action': 'set_password', 'current': 'wrong', 'new': 'pr3j5Dwi', 'confirm': 'pr3j5Dwi'},
            )
            self.assertStatus(200)
            self.assertInBody("Wrong current password.")
        # Then user session is cleared and user is redirect to login page
        self.getPage(
            '/prefs/general',
            method='POST',
            body={'action': 'set_password', 'current': 'wrong', 'new': 'pr3j5Dwi', 'confirm': 'pr3j5Dwi'},
        )
        self.assertStatus(303)
        self.assertHeaderItemValue('Location', self.baseurl + '/')
        # Then a warning message is displayed on login page
        self.getPage('/login/')
        self.assertStatus(200)
        default=20,
        help='maximum number of requests per hour that can be made on sensitive endpoints. When this limit is reached, an HTTP 429 message is returned to the user or the user is logged out. This security measure is used to limit brute force attacks on the login page and the RESTful API.',
                'tools.ratelimit.delay': 3600,
                'tools.ratelimit.limit': cfg.rate_limit,
# udb, A web interface to manage IT network
# Copyright (C) 2022 IKUS Software inc.
    def get_and_increment(self, token, delay, hit=1):
            tracker = tracker._replace(hits=tracker.hits + hit)
        except Exception:
def check_ratelimit(
    delay=3600, limit=25, return_status=429, logout=False, scope=None, methods=None, debug=False, hit=1, **conf
):
    Verify the ratelimit. By default return a 429 HTTP error code (Too Many Request). After 25 request within the same hour.

    Arguments:
        delay:         Time window for analysis in seconds. Default per hour (3600 seconds)
        limit:         Number of request allowed for an entry point. Default 25
        return_status: HTTP Error code to return.
        logout:        True to logout user when limit is reached
        scope:         if specify, define the scope of rate limit. Default to path_info.
        methods:       if specify, only the methods in the list will be rate limited.
    """
    assert delay > 0, 'invalid delay'
    # Check if limit is enabled
    if limit <= 0:
        return
    # Check if this 'method' should be rate limited
    request = cherrypy.request
    if methods is not None and request.method not in methods:
        if debug:
            cherrypy.log(
                'skip rate limit for HTTP method %s' % (request.method,),
                'TOOLS.RATELIMIT',
            )
        return
    datastore = getattr(cherrypy.request.app, '_ratelimit_datastore', None)
        cherrypy.request.app._ratelimit_datastore = datastore
    token = (request.login or request.remote.ip) + '.' + (scope or request.path_info)
    hits = datastore.get_and_increment(token, delay, hit)
            'check and increase rate limit for scope %s, limit %s, hits %s' % (token, limit, hits), 'TOOLS.RATELIMIT'
        if logout:
            if hasattr(cherrypy, 'session'):
                cherrypy.session.clear()
            raise cherrypy.HTTPRedirect("/")

        raise cherrypy.HTTPError(return_status)


def hit(hit=1):
    """
    May be called directly by handlers to add a hit for the given request.
    """
    conf = cherrypy.tools.ratelimit._merged_args()
    conf['hit'] = hit
    cherrypy.tools.ratelimit.callable(**conf)


cherrypy.tools.ratelimit.hit = hit