feat: ban a user after too many failed attempts

command.py

@@ -6,6 +6,7 @@ import pymumble_py3 as pymumble
 import re
 
 import constants
+import interface
 import media.system
 import util
 import variables as var
@@ -1181,18 +1182,28 @@ def cmd_web_access(bot, user, text, command, parameter):
     import datetime
     import json
 
-    user_info = var.db.get("user", user, fallback=None)
+    auth_method = var.config.get("webinterface", "auth_method")
-    if user_info is not None:
+
-        user_dict = json.loads(user_info)
+    if auth_method == 'token':
-        token = user_dict['token']
+        interface.banned_ip = []
+        interface.bad_access_count = {}
+
+        user_info = var.db.get("user", user, fallback=None)
+        if user_info is not None:
+            user_dict = json.loads(user_info)
+            token = user_dict['token']
+        else:
+            token = secrets.token_urlsafe(5)
+            var.db.set("web_token", token, user)
+
+        var.db.set("user", user, json.dumps({'token': token, 'datetime': str(datetime.datetime.now()), 'IP': ''}))
+
+        access_address = var.config.get("webinterface", "access_address") + "/?token=" + token
     else:
-        token = secrets.token_urlsafe(5)
+        access_address = var.config.get("webinterface", "access_address")
-        var.db.set("web_token", token, user)
 
-    var.db.set("user", user, json.dumps({'token': token, 'datetime': str(datetime.datetime.now()), 'IP': ''}))
+    bot.send_msg(constants.strings('webpage_address', address=access_address), text)
 
-    access_address = var.config.get("webinterface", "access_address")
-    bot.send_msg(constants.strings('webpage_token', address=access_address, token=token), text)
 
 # Just for debug use
 def cmd_real_time_rms(bot, user, text, command, parameter):

configuration.default.ini

@@ -98,7 +98,8 @@ web_logfile =
 
 auth_method = 'none'
 user = 
-password = 
+password =
+max_attempts = 10
 
 access_address = http://127.0.0.1:8181
 
@@ -261,7 +262,7 @@ cleared_tags = Removed all tags from <b>{song}</b>.
 cleared_tags_from_all = Removed all tags from songs on the playlist.
 shortlist_instruction = Use <i>!sl {indexes}</i> to play the item you want.
 auto_paused = Use <i>!play</i> to resume music!
-webpage_token= Your own address to access the web interface is <a href="{address}/?token={token}">{address}/?token={token}</a>
+webpage_address= Your own address to access the web interface is <a href="{address}">{address}</a>
 
 help = <h3>Commands</h3>
        <b>Control</b>

configuration.example.ini

@@ -128,8 +128,12 @@ port = 64738
 #web_logfile =
 
 # 'auth_method': Method used to authenticate users accessing the web interface.
-#                Options are 'none', 'password' or 'token' (use unique token, see requests_webinterface_access command)
+#                Options are 'none', 'password' or 'token' (use unique token, see
+#                requests_webinterface_access command)
+# 'max_attempts': Bad access attempts before being banned. Regenerating a token or
+#                 rebooting the bot will reset this attempts tally.
 #auth_method = token
+#max_attempts = 10
 
 # 'user', 'password': If auth_method set to 'password', you need to set the username and
 #                     password.

interface.py

@@ -93,10 +93,17 @@ def authenticate():
                     {'WWW-Authenticate': 'Basic realm="Login Required"'})
 
 
+bad_access_count = {}
+banned_ip = []
+
+
 def requires_auth(f):
     @wraps(f)
     def decorated(*args, **kwargs):
-        global log, user
+        global log, user, bad_access_count, banned_ip
+
+        if request.remote_addr in banned_ip:
+            abort(403)
 
         auth_method = var.config.get("webinterface", "auth_method")
 
@@ -104,7 +111,17 @@ def requires_auth(f):
             auth = request.authorization
             if not auth or not check_auth(auth.username, auth.password):
                 if auth:
-                    log.info(f"web: failed login attempt, user: {auth.username}, from ip {request.remote_addr}.")
+                    if request.remote_addr in bad_access_count:
+                        bad_access_count[request.remote_addr] += 1
+                        log.info(f"web: failed login attempt, user: {auth.username}, from ip {request.remote_addr}."
+                                 f"{bad_access_count[request.remote_addr]} attempts.")
+                        if bad_access_count[request.remote_addr] > var.config.getint("webinterface", "max_attempts",
+                                                                                     fallback=10):
+                            banned_ip.append(request.remote_addr)
+                            log.info(f"web: access banned for {request.remote_addr}")
+                    else:
+                        bad_access_count[request.remote_addr] = 1
+                        log.info(f"web: failed login attempt, user: {auth.username}, from ip {request.remote_addr}.")
                 return authenticate()
         if auth_method == 'token':
             if 'user' in session and 'token' not in request.args:
@@ -121,15 +138,26 @@ def requires_auth(f):
                     user_dict['IP'] = request.remote_addr
                     var.db.set("user", user, json.dumps(user_dict))
 
-                    log.debug(f"web: new user access, token validated for the user: {token_user}, from ip {request.remote_addr}.")
+                    log.debug(
+                        f"web: new user access, token validated for the user: {token_user}, from ip {request.remote_addr}.")
                     session['token'] = token
                     session['user'] = token_user
                     return f(*args, **kwargs)
 
-            log.info(f"web: bad token from ip {request.remote_addr}.")
+            if request.remote_addr in bad_access_count:
+                bad_access_count[request.remote_addr] += 1
+                log.info(f"web: bad token from ip {request.remote_addr}, "
+                         f"{bad_access_count[request.remote_addr]} attempts.")
+                if bad_access_count[request.remote_addr] > var.config.getint("webinterface", "max_attempts", fallback=10):
+                    banned_ip.append(request.remote_addr)
+                    log.info(f"web: access banned for {request.remote_addr}")
+            else:
+                bad_access_count[request.remote_addr] = 1
+                log.info(f"web: bad token from ip {request.remote_addr}.")
+
             return render_template('need_token.html',
-                                   name=var.config.get('bot','username'),
+                                   name=var.config.get('bot', 'username'),
-                                   command=f"{var.config.get('commands', 'command_symbol')[0]}{var.config.get('commands','requests_webinterface_access')}")
+                                   command=f"{var.config.get('commands', 'command_symbol')[0]}{var.config.get('commands', 'requests_webinterface_access')}")
 
         return f(*args, **kwargs)
 

static/js/custom.js

@@ -72,11 +72,14 @@ function request(_url, _data, refresh = false) {
                 }
                 updateControls(data.empty, data.play, data.mode, data.volume);
                 updatePlayerPlayhead(data.playhead);
+            },
+            403: function (){
+                location.reload(true);
             }
         },
     });
     if (refresh) {
-        location.reload()
+        location.reload(true)
     }
 }
 
@@ -541,6 +544,9 @@ function updateResults(dest_page = 1) {
                     lib_loading.hide();
                     lib_empty.show();
                     page_ul.empty();
+                },
+                403: function () {
+                    location.reload(true);
                 }
             }
         });