Implement mechanism for deactivating users

.github/workflows/.ci.yml

@@ -58,11 +58,8 @@ jobs:
           python -m pip install --upgrade pip
           python -m pip install .[test]
 
-      - name: Create environment file
-        run: cp ldap_jwt_auth/.env.example ldap_jwt_auth/.env
-
       - name: Create logging configuration file
         run: cp ldap_jwt_auth/logging.example.ini ldap_jwt_auth/logging.ini
 
       - name: Run unit tests
-        run: pytest test/unit/ --cov
+        run: pytest -c test/pytest.ini test/unit/ --cov

README.md

@@ -75,7 +75,6 @@ Ensure that you have an LDAP server to connect to.
 5. Create a `logging.ini` file alongside the `logging.example.ini` file. Use the example file as a reference and modify
    it accordingly.
 
-
 6. Create a `keys` directory in the root of the project directory, navigate to it, and generate OpenSSH encoded private and public key pair:
    ```bash
    mkdir keys
@@ -85,7 +84,6 @@ Ensure that you have an LDAP server to connect to.
 
 7. Create a `active_usernames.txt` file alongside the `active_usernames.example.txt` file and add all the usernames (each one on a seperate line) that are active/can access the system.
 
-
 8. Start the microservice using Uvicorn:
    ```bash
    uvicorn ldap_jwt_auth.main:app --log-config ldap_jwt_auth/logging.ini --reload
@@ -95,5 +93,5 @@ Ensure that you have an LDAP server to connect to.
 
 9. To run the unit tests, run:
    ```bash
-   pytest test/unit/
+   pytest -c test/pytest.ini test/unit/
    ```

ldap_jwt_auth/auth/jwt_handler.py

@@ -9,9 +9,10 @@
 import jwt
 from cryptography.hazmat.primitives import serialization
 
+from ldap_jwt_auth.auth.authentication import Authentication
 from ldap_jwt_auth.core.config import config
 from ldap_jwt_auth.core.constants import PRIVATE_KEY, PUBLIC_KEY
-from ldap_jwt_auth.core.exceptions import InvalidJWTError, JWTRefreshError
+from ldap_jwt_auth.core.exceptions import InvalidJWTError, JWTRefreshError, UserNotActiveError
 
 logger = logging.getLogger()
 
@@ -48,15 +49,25 @@ def get_refresh_token(self) -> str:
     def refresh_access_token(self, access_token: str, refresh_token: str):
         """
         Refreshes the JWT access token by updating its expiry time, provided that the JWT refresh token is valid.
+
+        Before attempting to refresh the token, it checks that the username is still part of the active usernames.
         :param access_token: The JWT access token to refresh.
         :param refresh_token: The JWT refresh token.
         :raises JWTRefreshError: If the JWT access token cannot be refreshed.
+        :raises UserNotActiveError: If the username is no longer part of the active usernames.
         :return: JWT access token with an updated expiry time.
         """
         logger.info("Refreshing access token")
         self.verify_token(refresh_token)
+
         try:
             payload = self._get_jwt_payload(access_token, {"verify_exp": False})
+
+            authentication = Authentication()
+            username = payload["username"]
+            if not authentication.is_user_active(username):
+                raise UserNotActiveError(f"The provided username '{username}' is not part of the active usernames")
+
             payload["exp"] = datetime.now(timezone.utc) + timedelta(
                 minutes=config.authentication.access_token_validity_minutes
             )

ldap_jwt_auth/routers/refresh.py

@@ -10,7 +10,7 @@
 from fastapi.responses import JSONResponse
 
 from ldap_jwt_auth.auth.jwt_handler import JWTHandler
-from ldap_jwt_auth.core.exceptions import JWTRefreshError, InvalidJWTError
+from ldap_jwt_auth.core.exceptions import JWTRefreshError, InvalidJWTError, ActiveUsernamesFileNotFoundError
 
 logger = logging.getLogger()
 
@@ -38,3 +38,7 @@ def refresh_access_token(
         message = "Unable to refresh access token"
         logger.exception(message)
         raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=message) from exc
+    except ActiveUsernamesFileNotFoundError as exc:
+        message = "Something went wrong"
+        logger.exception(message)
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=message) from exc

pyproject.toml

@@ -28,7 +28,8 @@ formatting = [
 
 test = [
     "pytest==8.0.0",
-    "pytest-cov==4.1.0"
+    "pytest-cov==4.1.0",
+    "pytest-env==1.1.3"
 ]
 
 dev = [

test/pytest.ini

@@ -0,0 +1,14 @@
+[pytest]
+env =
+    API__TITLE=LDAP-JWT Authentication Service API
+    API__DESCRIPTION=This is the API for the LDAP-JWT Authentication Service
+    # (If using a proxy) The path prefix handled by a proxy that is not seen by the app.
+    API__ROOT_PATH=
+    AUTHENTICATION__PRIVATE_KEY_PATH=./test/keys/jwt-key
+    AUTHENTICATION__PUBLIC_KEY_PATH=./test/keys/jwt-key.pub
+    AUTHENTICATION__JWT_ALGORITHM=RS256
+    AUTHENTICATION__ACCESS_TOKEN_VALIDITY_MINUTES=5
+    AUTHENTICATION__REFRESH_TOKEN_VALIDITY_DAYS=7
+    AUTHENTICATION__ACTIVE_USERNAMES_PATH=./test/active_usernames.txt
+    LDAP_SERVER__URL=ldap://ldap.example.com:389
+    LDAP_SERVER__REALM=LDAP.EXAMPLE.COM

test/unit/auth/test_jwt_handler.py

@@ -87,25 +87,44 @@ def test_get_refresh_token(datetime_mock):
     assert refresh_token == EXPECTED_REFRESH_TOKEN
 
 
+@patch("ldap_jwt_auth.auth.jwt_handler.Authentication.is_user_active")
 @patch("ldap_jwt_auth.auth.jwt_handler.datetime")
-def test_refresh_access_token(datetime_mock):
+def test_refresh_access_token(datetime_mock, is_user_active_mock):
     """
     Test refreshing an expired access token with a valid refresh token.
     """
     datetime_mock.now.return_value = mock_datetime_now()
+    is_user_active_mock.return_value = True
 
     jwt_handler = JWTHandler()
     access_token = jwt_handler.refresh_access_token(EXPIRED_ACCESS_TOKEN, VALID_REFRESH_TOKEN)
 
     assert access_token == EXPECTED_ACCESS_TOKEN
 
 
+@patch("ldap_jwt_auth.auth.jwt_handler.Authentication.is_user_active")
+def test_refresh_access_token_with_not_active_username(is_user_active_mock):
+    """
+    Test refreshing an access token when username is not active.
+    :param is_user_active_mock:
+    :return:
+    """
+    is_user_active_mock.return_value = False
+
+    jwt_handler = JWTHandler()
+    with pytest.raises(JWTRefreshError) as exc:
+        jwt_handler.refresh_access_token(EXPIRED_ACCESS_TOKEN, VALID_REFRESH_TOKEN)
+    assert str(exc.value) == "Unable to refresh access token"
+
+
+@patch("ldap_jwt_auth.auth.jwt_handler.Authentication.is_user_active")
 @patch("ldap_jwt_auth.auth.jwt_handler.datetime")
-def test_refresh_access_token_with_valid_access_token(datetime_mock):
+def test_refresh_access_token_with_valid_access_token(datetime_mock, is_user_active_mock):
     """
     Test refreshing a valid access token with a valid refresh token.
     """
     datetime_mock.now.return_value = mock_datetime_now()
+    is_user_active_mock.return_value = True
 
     jwt_handler = JWTHandler()
     access_token = jwt_handler.refresh_access_token(VALID_ACCESS_TOKEN, VALID_REFRESH_TOKEN)