feat: 2fa implementation

Author: fiqrikm18

src/core/security/two_factor_auth.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import base64
 import secrets
 from typing import Literal
 from uuid import UUID
@@ -11,14 +10,14 @@
 
 from src.core.config.setting import get_settings
 from src.core.email.service import EmailService
-from src.modules.user.domain.entities.user import User, UserSecurity
+from src.modules.user.domain.entities.user import User
 from src.modules.user.domain.repositories.user_repository import UserRepository
 from src.shared.unit_of_work import UnitOfWork
 
 
 class TwoFactorAuthService:
     """Service for managing two-factor authentication.
-    
+
     Supports:
     - TOTP (Time-based One-Time Password) for authenticator apps like Google Authenticator, Authy, etc.
     - Email-based 2FA codes
@@ -37,7 +36,7 @@ def __init__(
 
     async def setup_totp(self, user_id: UUID) -> dict[str, str]:
         """Set up TOTP for a user.
-        
+
         Returns:
             dict with 'secret', 'uri', and 'qr_code_data' keys
         """
@@ -48,14 +47,13 @@ async def setup_totp(self, user_id: UUID) -> dict[str, str]:
 
             # Generate a new secret
             secret = pyotp.random_base32()
-            
+
             # Create TOTP URI for QR code generation
             issuer = self._settings.JWT_ISSUER or "TodoApp"
             uri = pyotp.totp.TOTP(secret).provisioning_uri(
-                name=user.email,
-                issuer_name=issuer
+                name=user.email, issuer_name=issuer
             )
-            
+
             # Store the secret temporarily (not enabled yet)
             user.security.two_factor_secret = secret
             user.security.two_factor_enabled = False
@@ -65,16 +63,16 @@ async def setup_totp(self, user_id: UUID) -> dict[str, str]:
             return {
                 "secret": secret,
                 "uri": uri,
-                "qr_code_data": f"otpauth://totp/{issuer}:{user.email}?secret={secret}&issuer={issuer}"
+                "qr_code_data": f"otpauth://totp/{issuer}:{user.email}?secret={secret}&issuer={issuer}",
             }
 
     async def verify_totp_setup(self, user_id: UUID, code: str) -> dict[str, list[str]]:
         """Verify TOTP setup and enable 2FA.
-        
+
         Args:
             user_id: The user's ID
             code: The TOTP code from the authenticator app
-            
+
         Returns:
             dict with 'backup_codes' key containing recovery codes
         """
@@ -93,7 +91,7 @@ async def verify_totp_setup(self, user_id: UUID, code: str) -> dict[str, list[st
 
             # Generate backup codes
             backup_codes = [secrets.token_hex(4) for _ in range(10)]
-            
+
             # Enable 2FA and store backup codes
             user.security.two_factor_enabled = True
             user.security.two_factor_backup_codes = ",".join(backup_codes)
@@ -104,11 +102,11 @@ async def verify_totp_setup(self, user_id: UUID, code: str) -> dict[str, list[st
 
     async def disable_totp(self, user_id: UUID, code: str) -> bool:
         """Disable TOTP 2FA for a user.
-        
+
         Args:
             user_id: The user's ID
             code: Current TOTP code or backup code for verification
-            
+
         Returns:
             True if successfully disabled
         """
@@ -122,7 +120,7 @@ async def disable_totp(self, user_id: UUID, code: str) -> bool:
 
             # Verify code
             verified = False
-            
+
             # Check if it's a backup code
             if user.security.two_factor_backup_codes:
                 backup_codes = user.security.two_factor_backup_codes.split(",")
@@ -151,10 +149,10 @@ async def disable_totp(self, user_id: UUID, code: str) -> bool:
 
     async def send_email_2fa_code(self, user_id: UUID) -> bool:
         """Send a 2FA code via email.
-        
+
         Args:
             user_id: The user's ID
-            
+
         Returns:
             True if email was sent successfully
         """
@@ -168,14 +166,15 @@ async def send_email_2fa_code(self, user_id: UUID) -> bool:
 
             # Generate a 6-digit code
             code = secrets.token_hex(3)[:6]
-            
+
             # Store the code temporarily in security settings (with expiry info)
             # In production, you'd want to store this in Redis with TTL
             if not user.security:
                 raise ValueError("User security not found")
-            
+
             # Store code with timestamp (format: "code:timestamp")
             import time
+
             user.security.two_factor_secret = f"email_code:{code}:{int(time.time())}"
             await self._user_repository.save_security(user.security)
             await self._unit_of_work.commit()
@@ -191,22 +190,22 @@ async def send_email_2fa_code(self, user_id: UUID) -> bool:
                 </body>
             </html>
             """
-            
+
             await self._email_service.send_email(
                 to=user.email,
                 subject="Your Verification Code",
                 html_body=html_body,
             )
-            
+
             return True
 
     async def verify_email_2fa_code(self, user_id: UUID, code: str) -> bool:
         """Verify an email-based 2FA code.
-        
+
         Args:
             user_id: The user's ID
             code: The code received via email
-            
+
         Returns:
             True if code is valid
         """
@@ -222,17 +221,18 @@ async def verify_email_2fa_code(self, user_id: UUID, code: str) -> bool:
             parts = stored.split(":")
             if len(parts) != 3:
                 raise ValueError("Invalid code format")
-            
+
             stored_code = parts[1]
             timestamp = int(parts[2])
-            
+
             import time
+
             current_time = int(time.time())
-            
+
             # Code expires after 10 minutes
             if current_time - timestamp > 600:
                 raise ValueError("Code has expired")
-            
+
             if stored_code != code:
                 raise ValueError("Invalid code")
 
@@ -244,35 +244,35 @@ async def verify_email_2fa_code(self, user_id: UUID, code: str) -> bool:
             return True
 
     async def verify_2fa_code(
-        self,
-        user: User,
-        code: str,
-        method: Literal["totp", "email", "backup"] = "totp"
+        self, user: User, code: str, method: Literal["totp", "email", "backup"] = "totp"
     ) -> bool:
         """Verify a 2FA code using the specified method.
-        
+
         Args:
             user: The user entity
             code: The verification code
             method: The verification method ('totp', 'email', or 'backup')
-            
+
         Returns:
             True if verification successful
         """
         if not user.security:
             raise ValueError("User security not found")
 
         if method == "totp":
-            if not user.security.two_factor_secret or not user.security.two_factor_enabled:
+            if (
+                not user.security.two_factor_secret
+                or not user.security.two_factor_enabled
+            ):
                 raise ValueError("TOTP 2FA is not enabled")
-            
+
             totp = pyotp.TOTP(user.security.two_factor_secret)
             return bool(totp.verify(code, valid_window=1))
 
         elif method == "backup":
             if not user.security.two_factor_backup_codes:
                 raise ValueError("No backup codes available")
-            
+
             backup_codes = user.security.two_factor_backup_codes.split(",")
             if code in backup_codes:
                 # Remove used backup code
@@ -289,13 +289,15 @@ async def verify_2fa_code(
 
         return False
 
-    async def regenerate_backup_codes(self, user_id: UUID, verify_code: str) -> dict[str, list[str]]:
+    async def regenerate_backup_codes(
+        self, user_id: UUID, verify_code: str
+    ) -> dict[str, list[str]]:
         """Regenerate backup codes for a user.
-        
+
         Args:
             user_id: The user's ID
             verify_code: Current TOTP code for verification
-            
+
         Returns:
             dict with 'backup_codes' key containing new recovery codes
         """

src/modules/user/presentation/dependency.py

@@ -2,7 +2,8 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from src.core.database.postgres.session import get_db, get_unit_of_work
-from src.core.email.factory import get_email_service
+from src.core.email.factory import create_email_service
+from src.core.email.service import EmailService
 from src.core.security.account_lockout import AccountLockoutService
 from src.core.security.audit import AuditService
 from src.core.security.infrastructure.repositories.audit_log_repository import (
@@ -53,6 +54,10 @@ def get_user_repository(db: AsyncSession = Depends(get_db)) -> UserRepository:
     return SQLAlchemyUserRepository(db)
 
 
+def get_email_service() -> EmailService:
+    return create_email_service()
+
+
 def get_refresh_token_repository(
     db: AsyncSession = Depends(get_db),
 ) -> RefreshTokenRepository:
@@ -128,6 +133,7 @@ def get_logout_handler(
 
 # Two-Factor Authentication Handlers
 
+
 def get_setup_totp_handler(
     user_repo: UserRepository = Depends(get_user_repository),
     unit_of_work: UnitOfWork = Depends(get_unit_of_work),
@@ -152,7 +158,7 @@ def get_disable_totp_handler(
 def get_send_email_2fa_code_handler(
     user_repo: UserRepository = Depends(get_user_repository),
     unit_of_work: UnitOfWork = Depends(get_unit_of_work),
-    email_service = Depends(get_email_service),
+    email_service=Depends(get_email_service),
 ) -> SendEmail2FACodeHandler:
     return SendEmail2FACodeHandler(user_repo, unit_of_work, email_service)