Implement IP blacklisting

Author: manisandro

schemas/qwc-db-auth.json

@@ -123,6 +123,14 @@
         "totp_issuer_name": {
           "description": "Issuer name for QR code URI. Default: 'QWC Services'",
           "type": "string"
+        },
+        "ip_blacklist_duration": {
+          "description": "How many seconds an IP will remain in the blacklist. See also ip_blacklist_max_attempt_count. Default: 300",
+          "type": "integer"
+        },
+        "ip_blacklist_max_attempt_count": {
+          "description": "After how many failed login attempts an IP will be blacklisted. Should be less than max_login_attempts. See also ip_blacklist_duration. Default: 10",
+          "type": "integer"
         }
       },
       "required": [

src/db_auth.py

@@ -16,12 +16,14 @@
 from werkzeug.security import check_password_hash
 
 from qwc_services_core.auth import get_username
+from qwc_services_core.cache import ExpiringDict
 from qwc_services_core.database import DatabaseEngine
 from qwc_services_core.config_models import ConfigModels
 from qwc_services_core.runtime_config import RuntimeConfig
 
 from forms import LoginForm, NewPasswordForm, EditPasswordForm, VerifyForm
 
+ip_blacklist = ExpiringDict()
 
 class DBAuth:
     """DBAuth class
@@ -92,6 +94,8 @@ def __init__(self, tenant, mail, app):
         self.totp_enabled = config.get('totp_enabled', False)
         self.totp_enabled_for_admin = config.get('totp_enabled_for_admin', False)
         self.totp_issuer_name = config.get('totp_issuer_name', 'QWC Services')
+        self.ip_blacklist_duration = config.get('ip_blacklist_duration', 300)
+        self.ip_blacklist_max_attempt_count = config.get('ip_blacklist_max_attempt_count', 10)
 
         db_engine = DatabaseEngine()
         self.config_models = ConfigModels(
@@ -700,6 +704,14 @@ def __user_is_authorized(self, user, password):
         :param User user: User instance
         :param str password: Password
         """
+        # Check if IP blacklisted
+        if self.ip_blacklist_duration > 0:
+            entry = ip_blacklist.lookup(request.remote_addr)
+            count = entry['value'] if entry else 0
+            if count >= self.ip_blacklist_max_attempt_count:
+                self.logger.info("IP %s is blacklisted with %s attempts" % (request.remote_addr, count))
+                return False, i18n.t('auth.ip_blacklisted')
+
         if user is None or user.password_hash is None:
             # invalid username or no password set
             return False, i18n.t('auth.auth_failed')
@@ -719,6 +731,13 @@ def __user_is_authorized(self, user, password):
         else:
             # invalid password
 
+            # add to ip blacklist
+            if self.ip_blacklist_duration > 0:
+                entry = ip_blacklist.lookup(request.remote_addr)
+                count = entry['value'] if entry else 0
+                ip_blacklist.set(request.remote_addr, count + 1, self.ip_blacklist_duration)
+                self.logger.info("Attempt count for IP %s: %s" % (request.remote_addr, count + 1))
+
             # increase failed login attempts counter
             user.failed_sign_in_count += 1
 

src/translations/auth.de.json

@@ -9,6 +9,7 @@
     "auth_failed": "Ungültiger Benutzername oder ungültiges Passwort",
     "account_locked": "Konto gesperrt",
     "validation_error": "Überprüfung des Formulars fehlgeschlagen",
+    "ip_blacklisted": "IP-Addresse ist vorübergehend blockiert",
 
     "login_link": "Anmeldung",
 

src/translations/auth.en.json

@@ -9,6 +9,7 @@
     "auth_failed": "Invalid username or password",
     "account_locked": "Account locked",
     "validation_error": "Form validation error",
+    "ip_blacklisted": "IP address is temporarily blacklisted",
 
     "login_link": "Sign In",
 

src/translations/auth.fr.json

@@ -9,6 +9,7 @@
     "auth_failed": "Identifiant ou mot de passe invalide",
     "account_locked": "Compte bloqué",
     "validation_error": "Erreur dans le formulaire",
+    "ip_blacklisted": "Adresse IP temporairement bloquée",
 
     "login_link": "Se connecter",