Merge pull request #20 from flavien-hugs/develop

Feat token to blacklist

.env.example

@@ -8,6 +8,7 @@ FULLNAME_MIN_LENGTH=<ChangeMe>
 PASSWORD_MIN_LENGTH=<ChangeMe>
 APP_DEFAULT_PORT=<ChangeMe>
 DEFAULT_PAGIGNIATE_PAGE_SIZE=<ChangeMe>
+BLACKLIST_TOKEN_FILE=<ChangeMe>
 
 DEFAULT_ADMIN_FULLNAME=<ChangeMe>
 DEFAULT_ADMIN_EMAIL=<ChangeMe>

.gitignore

@@ -172,3 +172,4 @@ k8s/secrets.yml
 notebook.ipynb
 dotenv/
 .ruff_cache
+.tokens.txt

pyproject.toml

@@ -67,7 +67,7 @@ exclude = '''
 '''
 
 [tool.pytest.ini_options]
-env_files = 'tests/test.env'
+env_files = 'tests/.test.env'
 env_override_existing_values = 1
 capture = "no"
 log-cli-level = "INFO"

src/__init__.py

@@ -21,6 +21,7 @@
 from src.models import Role, User
 from src.routers import auth_router, perm_router, role_router, user_router
 from src.services import roles, users
+from src.shared import blacklist_token
 
 BASE_URL = slugify(settings.APP_NAME)
 
@@ -39,6 +40,8 @@ async def lifespan(app: FastAPI) -> AsyncIterator[State]:
     await roles.create_first_role()
     await users.create_first_user()
 
+    blacklist_token.init_blacklist_token_file()
+
     yield
     await shutdown_db(app=app)
 

src/middleware/auth.py

@@ -14,6 +14,7 @@
 from src.config import jwt_settings
 from src.services.roles import get_one_role
 from src.shared.error_codes import AuthErrorCode
+from src.shared import blacklist_token
 
 logging.basicConfig(format="%(message)s", level=logging.INFO)
 
@@ -79,6 +80,12 @@ def decode_access_token(cls, token: str) -> dict:
     @classmethod
     async def verify_access_token(cls, token: str) -> bool:
         try:
+            if await blacklist_token.is_token_blacklisted(token):
+                raise CustomHTTException(
+                    code_error=AuthErrorCode.AUTH_EXPIRED_ACCESS_TOKEN,
+                    message_error="Token has expired !",
+                    status_code=status.HTTP_401_UNAUTHORIZED,
+                )
             decode_token = cls.decode_access_token(token)
             current_timestamp = datetime.now(timezone.utc).timestamp()
 

src/services/auth.py

@@ -12,7 +12,7 @@
 from src.middleware.auth import CustomAccessBearer
 from src.models import User
 from src.schemas import ChangePassword, LoginUser
-from src.shared import mail_service
+from src.shared import mail_service, blacklist_token
 from src.shared.error_codes import AuthErrorCode, UserErrorCode
 from src.shared.utils import password_hash, verify_password
 from .roles import get_one_role
@@ -57,7 +57,10 @@ async def login(payload: LoginUser) -> JSONResponse:
 
 
 async def logout(request: Request) -> JSONResponse:
+
     authorization = request.headers.get("Authorization").split()[1]
+    await blacklist_token.add_blacklist_token(authorization)
+
     decode_token = CustomAccessBearer.decode_access_token(authorization)
 
     data = decode_token["subject"]

src/shared/__init__.py

@@ -1,5 +1,7 @@
 from .send_email import get_mail_service  # noqa: F401
+from .utils import TokenBlacklistHandler
 
 mail_service = get_mail_service()
+blacklist_token = TokenBlacklistHandler()
 
-__all__ = [mail_service]
+__all__ = ["mail_service", "blacklist_token"]

src/shared/utils.py

@@ -1,19 +1,25 @@
+import os
+import logging
 from enum import StrEnum
+from secrets import compare_digest
 
+import pyotp
 from fastapi_pagination import Page
 from fastapi_pagination.customization import CustomizedPage, UseParamsFields
 from fastapi_pagination.utils import disable_installed_extensions_check
 from pwdlib import PasswordHash
 from pwdlib.hashers.argon2 import Argon2Hasher
 from pwdlib.hashers.bcrypt import BcryptHasher
 
-import pyotp
 from src.config import settings
 
 disable_installed_extensions_check()
 
 password_context = PasswordHash((Argon2Hasher(), BcryptHasher()))
 
+logging.basicConfig(format="%(message)s", level=logging.INFO)
+logger = logging.getLogger(__name__)
+
 
 class SortEnum(StrEnum):
     ASC = "asc"
@@ -45,3 +51,40 @@ def generate_otp_instance(cls, key: str) -> pyotp.TOTP:
     @classmethod
     def verify_opt_code(cls, secret_otp: str, verify_otp: str) -> bool:
         return cls.generate_otp_instance(secret_otp).verify(verify_otp)
+
+
+class TokenBlacklistHandler:
+
+    def __init__(self):
+        self._token_file = os.getenv("BLACKLIST_TOKEN_FILE")
+        if not self._token_file:
+            raise ValueError("Blacklist file does not exist !")
+        self.init_blacklist_token_file()
+
+    def init_blacklist_token_file(self) -> bool:
+        try:
+            open(file=self._token_file, mode="a").close()
+            logger.info("Initialising the token blacklist file !")
+        except IOError as e:
+            raise IOError(f"Error when initialising the token blacklist file: {e}") from e
+        return True
+
+    async def add_blacklist_token(self, token: str) -> bool:
+        try:
+            with open(file=self._token_file, mode="a") as file:
+                file.write(f"{token},")
+                logger.info("Adding token to blacklist file !")
+        except IOError as e:
+            raise IOError(f"Error when adding token to blacklist: {e}") from e
+        return True
+
+    async def is_token_blacklisted(self, token: str) -> bool:
+        try:
+            with open(file=self._token_file) as file:
+                content = file.read()
+                tokens = content.rstrip(",").split(",")
+                logger.info("The token already exists in the blacklist !")
+        except IOError as e:
+            raise IOError(f"Error verifying token in black list: {e}") from e
+
+        return any(compare_digest(value, token) for value in tokens)

tests/test.env → tests/.test.env

@@ -11,6 +11,7 @@ DEFAULT_PAGIGNIATE_PAGE_SIZE=10
 FRONTEND_URL=http://${APP_HOSTNAME}:${APP_DEFAULT_PORT}
 FRONTEND_PATH_RESET_PASSWORD="/app/confirm/?token="
 FRONTEND_PATH_LOGIN="/connexion"
+BLACKLIST_TOKEN_FILE='.tokens.txt'
 
 DEFAULT_ADMIN_FULLNAME="Admin HAF"
 DEFAULT_ADMIN_EMAIL=test@localhost.com