Dom pwd (#132)

* Fixed #127

Author: terazus

ptmd/api/queries/users.py

@@ -46,6 +46,10 @@ def create_user() -> tuple[Response, int]:
         return jsonify(user_dict), 200
     except IntegrityError:
         return jsonify({"msg": "Username or email already taken"}), 400
+    except PasswordPolicyError as e:
+        return jsonify({"msg": str(e)}), 400
+    except Exception:
+        return jsonify({"msg": "An unexpected error occurred"}), 500
 
 
 def login() -> tuple[Response, int]:

ptmd/database/const.py

@@ -7,4 +7,4 @@
 
 SQLALCHEMY_DATABASE_URI: str = DOT_ENV_CONFIG['SQLALCHEMY_DATABASE_URL']
 SQLALCHEMY_SECRET_KEY: str = DOT_ENV_CONFIG['SQLALCHEMY_SECRET_KEY']
-PASSWORD_POLICY: str = "^(?=.*?[A-Z])(?=.*?[a-z])(?=.*?[0-9])(?=.*?[#?!@$%^&*-]).{8,20}$"
+PASSWORD_POLICY: str = r"^(?=.*?[A-Z])(?=.*?[a-z])(?=.*?[0-9])(?=.*?[\[\]\(\)#?!@$%^&*-_+=<>:;,.]).{8,20}$"

ptmd/database/models/user.py

@@ -48,6 +48,8 @@ def __init__(
     ) -> None:
         """ Constructor for the User class. Let's use encode the password with bcrypt before committing it to the
         database. """
+        if not match(PASSWORD_POLICY, password):
+            raise PasswordPolicyError
         self.username = username
         self.password = bcrypt.hash(password)
         self.email = email
@@ -104,7 +106,7 @@ def set_password(self, password: str) -> None:
         :raises PasswordPolicyError: if the password does not match the password policy
         """
         if not match(PASSWORD_POLICY, password):
-            raise PasswordPolicyError()
+            raise PasswordPolicyError
         self.password = bcrypt.hash(password)
         session.commit()
 

ptmd/database/queries/users.py

@@ -55,5 +55,6 @@ def get_token(token: str) -> Token:
     if token_from_db is None:
         raise TokenInvalidError
     if token_from_db.expires_on < datetime.now(token_from_db.expires_on.tzinfo):
+        # session.delete(token_from_db)  # type: ignore
         raise TokenExpiredError
     return token_from_db

tests/test_api/test_queries/test_users.py

@@ -6,7 +6,7 @@
 from sqlalchemy.exc import IntegrityError
 
 from ptmd.api import app
-from ptmd.exceptions import PasswordPolicyError
+from ptmd.exceptions import PasswordPolicyError, TokenInvalidError, TokenExpiredError
 
 
 HEADERS = {'Content-Type': 'application/json'}
@@ -91,6 +91,33 @@ def test_create_user(self, mock_organisation, mock_user,
                                        data=dumps(user_data))
             self.assertEqual(created_user.json, {'msg': 'Username or email already taken'})
 
+    @patch('ptmd.api.queries.users.Organisation')
+    def test_create_user_invalid_password(
+            self, mock_organisation, mock_get_current_user, mock_verify_jwt, mock_verify_in_request):
+        mock_get_current_user().role = 'admin'
+        user_data = {
+            "username": "1234",
+            "password": "1234",
+            "confirm_password": "1234",
+            "organisation": "UOX",
+            "email": "[email protected]"
+        }
+        with app.test_client() as client:
+            response = client.post('/api/users', headers={'Authorization': f'Bearer {123}', **HEADERS},
+                                   data=dumps(user_data))
+            self.assertEqual(response.json, {'msg': 'Password must be between 8 and 20 characters long, contain at '
+                                                    'least one uppercase letter, one lowercase letter, one number '
+                                                    'and one special character.'})
+            self.assertEqual(response.status_code, 400)
+
+            user_data['password'] = '!@#$%a^&A()a'
+            user_data['confirm_password'] = '!@#$%a^&A()a'
+            mock_organisation.query.filter.side_effect = Exception
+            response = client.post('/api/users', headers={'Authorization': f'Bearer {123}', **HEADERS},
+                                   data=dumps(user_data))
+            self.assertEqual(response.json, {'msg': 'An unexpected error occurred'})
+            self.assertEqual(response.status_code, 500)
+
     @patch('ptmd.api.queries.users.session')
     @patch('ptmd.api.queries.users.get_jwt', return_value={'sub': 1})
     @patch('ptmd.api.queries.users.User')
@@ -283,17 +310,26 @@ def test_reset_password_failed(self, mock_get_current_user, mock_verify_jwt, moc
     @patch('ptmd.api.queries.users.get_token')
     def test_reset_password_error(self, mock_token,
                                   mock_get_current_user, mock_verify_jwt, mock_verify_in_request):
-        mock_token.side_effect = PasswordPolicyError()
-        headers = {'Authorization': f'Bearer {123}', **HEADERS}
+        mock_token.return_value.user_reset[0].set_password.side_effect = PasswordPolicyError
+        headers = {'Authorization': 'Bearer 123', **HEADERS}
         with app.test_client() as client:
-            response = client.post('/api/users/reset/123', data=dumps({"password": "None"}), headers=headers)
+            response = client.post('/api/users/reset/456', data=dumps({"password": "None"}), headers=headers)
             self.assertEqual(response.json, {"msg": "Password must be between 8 and 20 characters long, contain at "
                                                     "least one uppercase letter, one lowercase letter, one number "
                                                     "and one special character."})
             self.assertEqual(response.status_code, 400)
 
-        mock_token.side_effect = Exception()
-        with app.test_client() as client:
+            mock_token.side_effect = TokenInvalidError
+            response = client.post('/api/users/reset/123', data=dumps({"password": "None"}), headers=headers)
+            self.assertEqual(response.json, {"msg": "Invalid token"})
+            self.assertEqual(response.status_code, 400)
+
+            mock_token.side_effect = TokenExpiredError
+            response = client.post('/api/users/reset/123', data=dumps({"password": "None"}), headers=headers)
+            self.assertEqual(response.json, {"msg": "Token expired"})
+            self.assertEqual(response.status_code, 400)
+
+            mock_token.side_effect = Exception
             response = client.post('/api/users/reset/123', data=dumps({"password": "None"}), headers=headers)
             self.assertEqual(response.json, {"msg": "An unexpected error occurred"})
             self.assertEqual(response.status_code, 500)

tests/test_database/test_models/test_user.py

@@ -40,7 +40,7 @@ def test_user_admin(self):
         organisation = Organisation(name='123', gdrive_id='1')
         organisation.files = []
         organisation.id = 2
-        user = User(username='test', password='test', email='[email protected]', role='admin')
+        user = User(username='test', password='!Str0?nkPassw0rd', email='[email protected]', role='admin')
         user.organisation = organisation
         self.assertEqual(user.role, 'admin')
         self.assertEqual(dict(user)['files'], [])
@@ -52,7 +52,7 @@ def test_user_with_organisation(self, mock_send_mail, mock_create_access_token):
         user_input: dict = {
             'username': 'rw1',
             'organisation_id': organisation.organisation_id,
-            'password': 'test',
+            'password': '!Str0?nkPassw0rd',
             'email': '[email protected]'
         }
         user = User(**user_input)
@@ -63,15 +63,15 @@ def test_user_with_organisation(self, mock_send_mail, mock_create_access_token):
     @patch('ptmd.database.models.user.session')
     @patch('ptmd.database.models.token.send_confirmation_mail', return_value=True)
     def test_set_role_success(self, mock_send_confirmation_mail, mock_session):
-        user = User('test', 'test', 'test', 'disabled')
+        user = User('test', '!Str0?nkPassw0rd', 'test', 'disabled')
         user.set_role('banned')
         self.assertEqual(user.role, 'banned')
         mock_session.commit.assert_called_once()
 
     @patch('ptmd.database.models.user.session')
     @patch('ptmd.database.models.token.send_confirmation_mail', return_value=True)
     def test_set_role_invalid_role(self, mock_send_confirmation_mail, mock_session):
-        user = User('test', 'test', 'test', 'disabled')
+        user = User('test', '!Str0?nkPassw0rd', 'test', 'disabled')
         with self.assertRaises(ValueError) as context:
             user.set_role('invalid role')
         self.assertEqual(str(context.exception), "Invalid role: invalid role")
@@ -89,7 +89,7 @@ def test_user_serialisation_with_organisation(self, mock_organisation, mock_orga
         organisation = Organisation(name='123', gdrive_id='1')
         organisation.files = [file_1, file_2]
         organisation.id = 2
-        user = User(username='test', password='test', email='[email protected]', role='admin')
+        user = User(username='test', password='!Str0?nkPassw0rd', email='[email protected]', role='admin')
         user.organisation = organisation
         user.files = [file_1]
         files = dict(user)['files']
@@ -98,9 +98,17 @@ def test_user_serialisation_with_organisation(self, mock_organisation, mock_orga
 
     @patch('ptmd.database.models.user.session')
     def test_set_password_policy_failure(self, mock_session):
-        user = User(username='test', password='test', email='[email protected]', role='admin')
+        user = User(username='test', password='!Str0?nkPassw0rd[]()', email='[email protected]', role='admin')
         with self.assertRaises(PasswordPolicyError) as context:
             user.set_password('test')
         self.assertEqual(str(context.exception),
-                         "Password must be between 8 and 20 characters long, contain at least one uppercase letter, one "
-                         "lowercase letter, one number and one special character.")
+                         "Password must be between 8 and 20 characters long, contain at least one uppercase letter, "
+                         "one lowercase letter, one number and one special character.")
+
+    def test_create_user_with_invalid_password(self):
+        user = User(username='test', password=':AStr0nkP3Wd!!', email='[email protected]', role='admin')
+        with self.assertRaises(PasswordPolicyError) as context:
+            user.set_password('test')
+        self.assertEqual(str(context.exception),
+                         "Password must be between 8 and 20 characters long, contain at least one uppercase letter, one"
+                         " lowercase letter, one number and one special character.")

tests/test_lib/test_email/test_core.py

@@ -11,7 +11,7 @@
 class TestEmailCore(TestCase):
 
     def test_send_validation_mail(self, mock_build, mock_get_config, mock_credentials):
-        user = User(username='username', email='email', password='password')
+        user = User(username='username', email='email', password='!Str0?nkPassw0rd')
         response = send_validation_mail(user)
         self.assertIn('<h1> Hello, admin </h1>', response)
 

tests/test_lib/test_email/test_load_templates.py

@@ -21,7 +21,7 @@ def test_validated_email(self):
 
     def test_create_validation_mail_content(self):
         organisation = Organisation(name='ORGANISATION NAME')
-        user = User(username='USERNAME', password='test', email='EMAIL', role='admin',
+        user = User(username='USERNAME', password='!Str0?nkPassw0rd', email='EMAIL', role='admin',
                     organisation_id=organisation.organisation_id)
         data = create_validation_mail_content(user)
         self.assertIn('USERNAME', data)