Use passlib.apache.HtpasswdFile instead of bcrypt to handle the databse file

Author: d0tiKs

setup.py

@@ -36,7 +36,8 @@
         'redis',
         ],
       extras_requires={
-        'bcrypt': ['bcrypt']
+        'passlib': ['passlib==1.7.4']
+        'bcrypt': ['bcrypt==4.0.1']
       },
       zip_safe=False,
       entry_points={

test-requirements.txt

@@ -3,4 +3,5 @@ nose2
 six
 redis
 wrapt<=1.12.1;python_version<="3.4"
-bcrypt
+passlib==1.7.4
+bcrypt==4.0.1

tests/test_auth_plugins.py

@@ -2,7 +2,7 @@
 
 """ Unit tests for Authentication plugins"""
 
-from websockify.auth_plugins import BasicHTTPAuth, HtPasswdAuth, AuthenticationError
+from websockify.auth_plugins import BasicHTTPAuth, HtpasswdAuth, AuthenticationError
 import unittest
 import tempfile
 
@@ -28,7 +28,7 @@ def test_garbage_auth(self):
         headers = {'Authorization': 'Basic xxxxxxxxxxxxxxxxxxxxxxxxxxxx'}
         self.assertRaises(AuthenticationError, self.plugin.authenticate, headers, 'localhost', '1234')
 
-class HtPasswdAuthTestCase(unittest.TestCase):
+class HtpasswdAuthTestCase(unittest.TestCase):
 
 
     def setUp(self):
@@ -38,7 +38,7 @@ def setUp(self):
         self._temporary_htpasswd_file.write(file_content.encode('utf-8'))
         self._temporary_htpasswd_file.close()
 
-        self.plugin = HtPasswdAuth(self._temporary_htpasswd_file.name)
+        self.plugin = HtpasswdAuth(self._temporary_htpasswd_file.name)
 
     def test_no_auth(self):
         headers = {}

websockify/auth_plugins.py

@@ -1,5 +1,7 @@
-import bcrypt
-
+try:
+    from passlib.apache import HtpasswdFile
+except ImportError:
+    HtpasswdFile: None
 class BasePlugin():
     def __init__(self, src=None):
         self.source = src
@@ -25,10 +27,9 @@ def __init__(self, expected, actual):
         self.expected_origin = expected
         self.actual_origin = actual
 
-        super().__init__(
-            response_msg='Invalid Origin',
-            log_msg="Invalid Origin Header: Expected one of "
-                    "%s, got '%s'" % (expected, actual))
+        super().__init__(response_msg='Invalid Origin',
+                         log_msg="Invalid Origin Header: Expected one of "
+                         "%s, got '%s'" % (expected, actual))
 
 
 class BasicHTTPAuth():
@@ -78,21 +79,25 @@ def demand_auth(self):
         raise AuthenticationError(response_code=401,
                                   response_headers={'WWW-Authenticate': 'Basic realm="Websockify"'})
 
-class HtPasswdAuth(BasicHTTPAuth):
+class HtpasswdAuth(BasicHTTPAuth):
     """Verifies Basic Auth headers against a htpasswd database. Specify src as the path to the htpasswd file"""
 
     def __init__(self, src=None):
         self.src = src
+        if HtpasswdFile is None:
+            raise AuthenticationError(response_code=500, response_msg=f"Internal Server Error")
 
     def validate_creds(self, username, password):
         if self.src == None:
             return False
         try:
-            with open(self.src, 'r') as file:
-                for line in file:
-                    stored_user, stored_hash = line.strip().split(':', 1)
-                    if stored_user == username:
-                        return bcrypt.checkpw(password.encode('utf-8'), stored_hash.encode('utf-8'))
+            #TODO: Add a argument or config to change the HtpasswdFile scheme
+            htfile = HtpasswdFile(self.src, new=False, default_scheme="bcrypt", encoding="utf-8")
+            isvalid_hash = htfile.check_password(username, password)
+            if isvalid_hash == None:
+                #log user not found
+                raise AuthenticationError(response_code=403)
+            return isvalid_hash
         except (FileNotFoundError, PermissionError, OSError, ValueError) as e:
             #log error "%s: %s" % (type(e).__name__, e)
             raise AuthenticationError(response_code=500, response_msg=f"Internal Server Error")