Refactor auth under nginx.check_credentials()

core/admin/mailu/internal/nginx.py

@@ -19,6 +19,20 @@ STATUSES = {
     }),
 }
 
+def check_credentials(user, password, ip, protocol=None):
+    if not user or not user.enabled or (protocol == "imap" and not user.enable_imap) or (protocol == "pop3" and not user.enable_pop):
+        return False
+    is_ok = False
+    # All tokens are 32 characters hex lowercase
+    if len(password) == 32:
+        for token in user.tokens:
+            if (token.check_password(password) and
+                (not token.ip or token.ip == ip)):
+                    is_ok = True
+                    break
+    if not is_ok and user.check_password(password):
+        is_ok = True
+    return is_ok
 
 def handle_authentication(headers):
     """ Handle an HTTP nginx authentication request
@@ -47,23 +61,7 @@ def handle_authentication(headers):
         password = raw_password.encode("iso8859-1").decode("utf8")
         ip = urllib.parse.unquote(headers["Client-Ip"])
         user = models.User.query.get(user_email)
-        status = False
-        if user:
-            # All tokens are 32 characters hex lowercase
-            if len(password) == 32:
-                for token in user.tokens:
-                    if (token.check_password(password) and
-                        (not token.ip or token.ip == ip)):
-                            status = True
-                            break
-            if not status and user.check_password(password):
-                status = True
-            if status:
-                if protocol == "imap" and not user.enable_imap:
-                    status = False
-                elif protocol == "pop3" and not user.enable_pop:
-                    status = False
-        if status and user.enabled:
+        if check_credentials(user, password, ip, protocol):
             return {
                 "Auth-Status": "OK",
                 "Auth-Server": server,

core/admin/mailu/internal/views/auth.py

@@ -53,22 +53,10 @@ def basic_authentication():
         encoded = authorization.replace("Basic ", "")
         user_email, password = base64.b64decode(encoded).split(b":")
         user = models.User.query.get(user_email.decode("utf8"))
-        if user and user.enabled:
-            password = password.decode('utf-8')
-            status = False
-            # All tokens are 32 characters hex lowercase
-            if len(password) == 32:
-                for token in user.tokens:
-                    if (token.check_password(password) and
-                        (not token.ip or token.ip == ip)):
-                            status = True
-                            break
-            if not status and user.check_password(password):
-                status = True
-            if status:
-                response = flask.Response()
-                response.headers["X-User"] = user.email
-                return response
+        if nginx.check_credentials(user, password.decode('utf-8'), flask.request.remote_addr, "web"):
+            response = flask.Response()
+            response.headers["X-User"] = user.email
+            return response
     response = flask.Response(status=401)
     response.headers["WWW-Authenticate"] = 'Basic realm="Login Required"'
     return response