feat: add authentication, user management, and phase 1 auth workflows to account app

apps/account/admin/role.py

@@ -107,6 +107,7 @@ class RoleAdmin(ModelAdmin):
             "fields": (
                 ("can_access_admin_panel", "can_manage_users"),
                 ("can_ban_users", "can_view_analytics"),
+                ("require_2fa",),
             ),
         }),
         (_("Metadata"), {

apps/account/admin/user.py

@@ -158,6 +158,7 @@ class UserAdmin(ModelAdmin, BaseUserAdmin):
             "classes": ["tab"],
             "fields": (
                 ("is_active", "is_staff", "is_superuser"),
+                ("two_factor_enabled", "two_factor_method"),
                 "custom_permissions",
                 ("role", "user_type"),
                 "groups",

apps/account/management/commands/seed_roles.py

@@ -340,7 +340,7 @@ DEFAULT_ROLES = [
         "description": "Master administrative role with all permissions, total access control, and full platform authority.",
         "is_default": False,
         "is_system": True,
-        "permissions": {perm: (perm != "is_chat_muted") for perm in ALL_PERMISSION_FIELDS}
+        "permissions": {perm: (perm not in ("is_chat_muted", "require_2fa")) for perm in ALL_PERMISSION_FIELDS}
     },
 ]
 

apps/account/migrations/0007_role_require_2fa_and_user_2fa_fields.py

@@ -0,0 +1,58 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('account', '0006_user_managed_regions_user_regional_permissions'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='role',
+            name='require_2fa',
+            field=models.BooleanField(
+                default=False,
+                help_text='If enabled, users with this role cannot log in without completing Two-Factor Authentication (2FA) and cannot disable 2FA.',
+                verbose_name='Force Two-Factor Authentication (2FA)',
+            ),
+        ),
+        migrations.AddField(
+            model_name='user',
+            name='two_factor_enabled',
+            field=models.BooleanField(
+                default=False,
+                help_text='Designates whether the user has enabled Two-Factor Authentication (2FA) for login.',
+                verbose_name='Two-Factor Authentication Enabled',
+            ),
+        ),
+        migrations.AddField(
+            model_name='user',
+            name='two_factor_method',
+            field=models.CharField(
+                choices=[('email', 'Email OTP'), ('totp', 'Authenticator App (TOTP)')],
+                default='email',
+                max_length=16,
+                verbose_name='Preferred 2FA Method',
+            ),
+        ),
+        migrations.AddField(
+            model_name='user',
+            name='totp_secret_encrypted',
+            field=models.TextField(
+                blank=True,
+                default='',
+                null=True,
+                verbose_name='Encrypted TOTP Secret',
+            ),
+        ),
+        migrations.AddField(
+            model_name='user',
+            name='two_factor_backup_codes',
+            field=models.JSONField(
+                blank=True,
+                default=list,
+                verbose_name='2FA Backup Recovery Codes (Hashed)',
+            ),
+        ),
+    ]

apps/account/models/role.py

@@ -73,6 +73,7 @@ PERMISSION_CATEGORIES = {
         "can_manage_users",
         "can_ban_users",
         "can_view_analytics",
+        "require_2fa",
     ],
 }
 
@@ -163,6 +164,11 @@ class Role(models.Model):
     can_manage_users = models.BooleanField(default=False, verbose_name=_("Manage Users"))
     can_ban_users = models.BooleanField(default=False, verbose_name=_("Ban / Suspend Users"))
     can_view_analytics = models.BooleanField(default=False, verbose_name=_("View Analytics Dashboard"))
+    require_2fa = models.BooleanField(
+        default=False,
+        verbose_name=_("Force Two-Factor Authentication (2FA)"),
+        help_text=_("If enabled, users with this role cannot log in without completing Two-Factor Authentication (2FA) and cannot disable 2FA.")
+    )
 
     created_at = models.DateTimeField(auto_now_add=True, verbose_name=_("Created At"))
     updated_at = models.DateTimeField(auto_now=True, verbose_name=_("Updated At"))

apps/account/models/user.py

@@ -167,12 +167,39 @@ class User(AbstractUser):
         verbose_name=_("Managed Regions"),
         help_text=_("Regions assigned to this regional administrator or staff member. Administrative actions and regional permissions apply only to these regions.")
     )
+    class TwoFactorMethod(models.TextChoices):
+        EMAIL = "email", _("Email OTP")
+        TOTP = "totp", _("Authenticator App (TOTP)")
+
     regional_permissions = models.JSONField(
         default=list,
         blank=True,
         verbose_name=_("Regional Permission Overrides"),
         help_text=_("List of permission names that are restricted to the user's managed regions (e.g. ['can_manage_institutions', 'can_manage_tickets']).")
     )
+
+    two_factor_enabled = models.BooleanField(
+        default=False,
+        verbose_name=_("Two-Factor Authentication Enabled"),
+        help_text=_("Designates whether the user has enabled Two-Factor Authentication (2FA) for login.")
+    )
+    two_factor_method = models.CharField(
+        max_length=16,
+        choices=TwoFactorMethod.choices,
+        default=TwoFactorMethod.EMAIL,
+        verbose_name=_("Preferred 2FA Method"),
+    )
+    totp_secret_encrypted = models.TextField(
+        blank=True,
+        null=True,
+        default='',
+        verbose_name=_("Encrypted TOTP Secret"),
+    )
+    two_factor_backup_codes = models.JSONField(
+        default=list,
+        blank=True,
+        verbose_name=_("2FA Backup Recovery Codes (Hashed)"),
+    )
     
     objects = UserManager()
      
@@ -252,6 +279,8 @@ class User(AbstractUser):
         """
         if not self.is_active:
             return False
+        if perm_name == 'require_2fa':
+            return self.is_2fa_forced_by_admin()
         if self.is_superuser:
             return True
         if self.custom_permissions and perm_name in self.custom_permissions:
@@ -422,6 +451,113 @@ class User(AbstractUser):
             return decrypted_bytes.decode('utf-8')
         except Exception:
             return ''
+
+    # ─── Two-Factor Authentication (2FA) Helpers ─────────────────────────────
+
+    def is_2fa_forced_by_admin(self) -> bool:
+        """
+        Returns True if an administrator has forced 2FA for this user (via custom_permissions['require_2fa'])
+        or for the user's assigned Role (Role.require_2fa).
+        """
+        if not self.is_active:
+            return False
+        if isinstance(self.custom_permissions, dict) and 'require_2fa' in self.custom_permissions:
+            val = self.custom_permissions['require_2fa']
+            if val is not None:
+                return bool(val)
+        if self.role and hasattr(self.role, 'require_2fa'):
+            return bool(self.role.require_2fa)
+        return False
+
+    def is_2fa_required(self) -> bool:
+        """
+        Returns True if this user must complete 2FA on login:
+        either because the user toggled 2FA on themselves, or because an admin forced it.
+        """
+        if not self.is_active:
+            return False
+        return bool(self.two_factor_enabled or self.is_2fa_forced_by_admin())
+
+    def has_totp_configured(self) -> bool:
+        return bool(self.get_totp_secret())
+
+    def get_effective_2fa_method(self) -> str:
+        """
+        Returns the active 2FA method ('totp' if TOTP is selected and secret is configured, otherwise 'email').
+        """
+        if self.two_factor_method == self.TwoFactorMethod.TOTP and self.has_totp_configured():
+            return self.TwoFactorMethod.TOTP
+        return self.TwoFactorMethod.EMAIL
+
+    def get_available_2fa_methods(self) -> list:
+        methods = [self.TwoFactorMethod.EMAIL]
+        if self.has_totp_configured():
+            methods.insert(0, self.TwoFactorMethod.TOTP)
+        if isinstance(self.two_factor_backup_codes, list) and len(self.two_factor_backup_codes) > 0:
+            methods.append("backup")
+        return methods
+
+    def set_totp_secret(self, raw_secret: str) -> None:
+        if not raw_secret:
+            self.totp_secret_encrypted = ''
+            return
+        try:
+            fernet = self._get_fernet()
+            self.totp_secret_encrypted = fernet.encrypt(raw_secret.encode('utf-8')).decode('utf-8')
+        except Exception:
+            self.totp_secret_encrypted = ''
+
+    def get_totp_secret(self) -> str:
+        if not self.totp_secret_encrypted:
+            return ''
+        try:
+            fernet = self._get_fernet()
+            return fernet.decrypt(self.totp_secret_encrypted.encode('utf-8')).decode('utf-8')
+        except Exception:
+            return ''
+
+    def verify_totp_code(self, code: str) -> bool:
+        secret = self.get_totp_secret()
+        if not secret:
+            return False
+        from utils.totp import verify_totp_code as _verify_totp
+        return _verify_totp(secret=secret, code=code)
+
+    def generate_backup_codes(self, count: int = 8) -> list:
+        """
+        Generates `count` single-use 8-character recovery codes, stores their SHA-256 hashes,
+        and returns the plaintext codes once.
+        """
+        import secrets as _secrets
+        alphabet = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789"
+        plaintext_codes = []
+        hashed_codes = []
+        for _ in range(count):
+            raw = "".join(_secrets.choice(alphabet) for _ in range(8))
+            formatted = f"{raw[:4]}-{raw[4:]}"
+            plaintext_codes.append(formatted)
+            normalized = raw.upper()
+            digest = hashlib.sha256(f"{settings.SECRET_KEY}:2fa_backup:{normalized}".encode("utf-8")).hexdigest()
+            hashed_codes.append(digest)
+        self.two_factor_backup_codes = hashed_codes
+        return plaintext_codes
+
+    def verify_and_consume_backup_code(self, code: str) -> bool:
+        """
+        Verifies a single-use backup recovery code and removes it upon match.
+        """
+        if not code or not isinstance(self.two_factor_backup_codes, list) or not self.two_factor_backup_codes:
+            return False
+        normalized = "".join(str(code).strip().split()).replace("-", "").upper()
+        if len(normalized) != 8:
+            return False
+        digest = hashlib.sha256(f"{settings.SECRET_KEY}:2fa_backup:{normalized}".encode("utf-8")).hexdigest()
+        if digest in self.two_factor_backup_codes:
+            remaining = [h for h in self.two_factor_backup_codes if h != digest]
+            self.two_factor_backup_codes = remaining
+            self.save(update_fields=["two_factor_backup_codes"])
+            return True
+        return False
         
     class Meta:
         ordering = ("-id",)

apps/account/serializers/auth_serializers.py

@@ -112,6 +112,12 @@ class UserMeSerializer(serializers.ModelSerializer):
     position = serializers.SerializerMethodField()
     institution = serializers.SerializerMethodField()
     institution_id = serializers.SerializerMethodField()
+    two_factor_enabled = serializers.SerializerMethodField()
+    two_factor_method = serializers.SerializerMethodField()
+    is_2fa_forced_by_admin = serializers.SerializerMethodField()
+    require_2fa = serializers.SerializerMethodField()
+    has_totp_configured = serializers.SerializerMethodField()
+    backup_codes_count = serializers.SerializerMethodField()
 
     class Meta:
         model = User
@@ -153,6 +159,12 @@ class UserMeSerializer(serializers.ModelSerializer):
             'regional_permissions',
             'verification_status',
             'is_verified',
+            'two_factor_enabled',
+            'two_factor_method',
+            'is_2fa_forced_by_admin',
+            'require_2fa',
+            'has_totp_configured',
+            'backup_codes_count',
         ]
         read_only_fields = [
             'id',
@@ -180,6 +192,12 @@ class UserMeSerializer(serializers.ModelSerializer):
             'regional_permissions',
             'verification_status',
             'is_verified',
+            'two_factor_enabled',
+            'two_factor_method',
+            'is_2fa_forced_by_admin',
+            'require_2fa',
+            'has_totp_configured',
+            'backup_codes_count',
         ]
 
     def get_first_name(self, obj) -> str:
@@ -278,10 +296,13 @@ class UserMeSerializer(serializers.ModelSerializer):
             return {}
         if getattr(obj, 'is_superuser', False):
             from apps.account.models.role import ALL_PERMISSION_FIELDS
-            return {perm: True for perm in ALL_PERMISSION_FIELDS}
+            flags = {perm: True for perm in ALL_PERMISSION_FIELDS}
+            flags['require_2fa'] = obj.is_2fa_forced_by_admin()
+            return flags
         flags = obj.role.get_permissions_dict() if obj.role else {}
         if obj.custom_permissions and isinstance(obj.custom_permissions, dict):
             flags.update(obj.custom_permissions)
+        flags['require_2fa'] = obj.is_2fa_forced_by_admin()
         return flags
 
     def get_permissions(self, obj) -> list:
@@ -291,16 +312,11 @@ class UserMeSerializer(serializers.ModelSerializer):
         """
         if not obj or not getattr(obj, 'is_active', False):
             return []
-        if getattr(obj, 'is_superuser', False):
-            from apps.account.models.role import ALL_PERMISSION_FIELDS
-            return list(ALL_PERMISSION_FIELDS)
-
-        flags = obj.role.get_permissions_dict() if obj.role else {}
-        if obj.custom_permissions and isinstance(obj.custom_permissions, dict):
-            flags.update(obj.custom_permissions)
-
+        flags = self.get_permission_flags(obj)
         granted = [k for k, v in flags.items() if v]
 
+        if obj.is_institution_admin and 'manage_institution_profile' not in granted:
+            granted.append('manage_institution_profile')
         if obj.is_super_admin and 'super_admin' not in granted:
             granted.append('super_admin')
         return granted
@@ -324,6 +340,26 @@ class UserMeSerializer(serializers.ModelSerializer):
             for r in obj.managed_regions.filter(is_active=True)
         ]
 
+    def get_two_factor_enabled(self, obj) -> bool:
+        return obj.is_2fa_required()
+
+    def get_two_factor_method(self, obj) -> str:
+        return obj.get_effective_2fa_method()
+
+    def get_is_2fa_forced_by_admin(self, obj) -> bool:
+        return obj.is_2fa_forced_by_admin()
+
+    def get_require_2fa(self, obj) -> bool:
+        return obj.is_2fa_forced_by_admin()
+
+    def get_has_totp_configured(self, obj) -> bool:
+        return obj.has_totp_configured()
+
+    def get_backup_codes_count(self, obj) -> int:
+        if isinstance(obj.two_factor_backup_codes, list):
+            return len(obj.two_factor_backup_codes)
+        return 0
+
 
 class UserMeUpdateSerializer(serializers.ModelSerializer):
     bio = serializers.CharField(source='info', allow_blank=True, required=False, allow_null=True)
@@ -339,6 +375,8 @@ class UserMeUpdateSerializer(serializers.ModelSerializer):
     institution_name = serializers.CharField(required=False, allow_blank=True)
     website = serializers.CharField(required=False, allow_blank=True)
     address = serializers.CharField(required=False, allow_blank=True)
+    two_factor_enabled = serializers.BooleanField(required=False)
+    two_factor_method = serializers.ChoiceField(choices=User.TwoFactorMethod.choices, required=False)
 
     class Meta:
         model = User
@@ -365,8 +403,26 @@ class UserMeUpdateSerializer(serializers.ModelSerializer):
             'city',
             'gender',
             'birthdate',
+            'two_factor_enabled',
+            'two_factor_method',
         ]
 
+    def validate(self, attrs):
+        attrs = super().validate(attrs)
+        instance = getattr(self, 'instance', None)
+        if instance and 'two_factor_enabled' in attrs and attrs['two_factor_enabled'] is False:
+            if instance.is_2fa_forced_by_admin():
+                from rest_framework.exceptions import PermissionDenied
+                raise PermissionDenied(
+                    _("Two-Factor Authentication (2FA) is enforced by an administrator and cannot be disabled.")
+                )
+        if instance and attrs.get('two_factor_method') == User.TwoFactorMethod.TOTP:
+            if not instance.has_totp_configured():
+                raise serializers.ValidationError({
+                    "two_factor_method": _("Please complete Authenticator App (TOTP) setup before selecting TOTP as your active method.")
+                })
+        return attrs
+
     def to_internal_value(self, data):
         if hasattr(data, 'copy'):
             mutable_data = data.copy()
@@ -499,9 +555,21 @@ class PasswordResetRequestSerializer(serializers.Serializer):
         return value
 
 
+def mask_email_address(email: str) -> str:
+    if not email or "@" not in email:
+        return email or ""
+    local, domain = email.split("@", 1)
+    if len(local) <= 2:
+        masked_local = local[0] + "*"
+    else:
+        masked_local = local[0] + ("*" * min(len(local) - 2, 5)) + local[-1]
+    return f"{masked_local}@{domain}"
+
+
 class CustomTokenObtainPairSerializer(TokenObtainPairSerializer):
     """
     JWT token obtain serializer that accepts either 'username' or 'email' as the credential.
+    Enforces Two-Factor Authentication (2FA) when user.is_2fa_required() is True.
     """
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
@@ -526,5 +594,94 @@ class CustomTokenObtainPairSerializer(TokenObtainPairSerializer):
             raise serializers.ValidationError({"email": _("This field is required.")})
         attrs['email'] = identifier
         attrs['username'] = identifier
-        return super().validate(attrs)
+        data = super().validate(attrs)
+
+        if self.user and self.user.is_2fa_required():
+            from utils.redis import RedisManager
+            from apps.account.tasks import dispatch_2fa_email_otp
+
+            redis_mgr = RedisManager()
+            effective_method = self.user.get_effective_2fa_method()
+            temp_token = redis_mgr.create_2fa_challenge(
+                user_id=self.user.id,
+                method=effective_method,
+                ttl=300,
+            )
+            if effective_method == User.TwoFactorMethod.EMAIL and self.user.email:
+                otp_code = redis_mgr.generate_2fa_otp_code()
+                redis_mgr.set_2fa_email_otp(temp_token, otp_code, ttl=300)
+                dispatch_2fa_email_otp(self.user.email, otp_code)
+
+            return {
+                "requires_2fa": True,
+                "temp_token": temp_token,
+                "method": effective_method,
+                "available_methods": self.user.get_available_2fa_methods(),
+                "email": self.user.email or "",
+                "masked_email": mask_email_address(self.user.email or ""),
+                "is_forced_by_admin": self.user.is_2fa_forced_by_admin(),
+                "is_2fa_forced_by_admin": self.user.is_2fa_forced_by_admin(),
+                "message": _("Two-Factor Authentication (2FA) is required to complete login."),
+            }
+
+        return data
+
+
+# ─── Two-Factor Authentication (2FA) Serializers ─────────────────────────────
+
+
+class TwoFactorVerifySerializer(serializers.Serializer):
+    temp_token = serializers.CharField(required=True, help_text="Ephemeral 2FA challenge token returned by /api/v1/auth/token/.")
+    code = serializers.CharField(required=True, min_length=6, max_length=16, help_text="6-digit OTP code (Email or TOTP) or 8-character backup recovery code.")
+    method = serializers.ChoiceField(
+        choices=[
+            ("email", "Email OTP"),
+            ("totp", "Authenticator App (TOTP)"),
+            ("backup", "Backup Recovery Code"),
+            ("backup_code", "Backup Recovery Code"),
+        ],
+        required=False,
+        help_text="Verification method used. Defaults to the challenge's active method.",
+    )
+
+
+class TwoFactorResendEmailSerializer(serializers.Serializer):
+    temp_token = serializers.CharField(required=True, help_text="Ephemeral 2FA challenge token.")
+
+
+class TwoFactorToggleSerializer(serializers.Serializer):
+    enabled = serializers.BooleanField(required=True, help_text="Set True to enable 2FA or False to disable 2FA.")
+    method = serializers.ChoiceField(
+        choices=User.TwoFactorMethod.choices,
+        required=False,
+        help_text="Optional preferred 2FA method ('email' or 'totp').",
+    )
+
+
+class TwoFactorSetupInitiateSerializer(serializers.Serializer):
+    method = serializers.ChoiceField(
+        choices=User.TwoFactorMethod.choices,
+        default=User.TwoFactorMethod.TOTP,
+        required=False,
+        help_text="2FA method to set up ('totp' or 'email').",
+    )
+
+
+class TwoFactorSetupConfirmSerializer(serializers.Serializer):
+    method = serializers.ChoiceField(
+        choices=User.TwoFactorMethod.choices,
+        default=User.TwoFactorMethod.TOTP,
+        required=False,
+        help_text="2FA method being confirmed ('totp' or 'email').",
+    )
+    code = serializers.CharField(required=True, min_length=6, max_length=8, help_text="6-digit verification code from Authenticator app or Email.")
+
+
+class TwoFactorDisableSerializer(serializers.Serializer):
+    password = serializers.CharField(required=False, allow_blank=True, write_only=True, help_text="Optional current password confirmation.")
+
+
+class TwoFactorBackupCodesRegenerateSerializer(serializers.Serializer):
+    password = serializers.CharField(required=False, allow_blank=True, write_only=True, help_text="Optional current password confirmation.")
+
 

apps/account/serializers/user.py

@@ -290,6 +290,19 @@ class AdminUserSerializer(serializers.ModelSerializer):
     def get_device_id(self, obj):
         return obj.device_id or None
 
+    require_2fa = serializers.BooleanField(required=False, default=False)
+    two_factor_enabled = serializers.BooleanField(required=False)
+
+    def to_representation(self, instance):
+        rep = super().to_representation(instance)
+        forced = instance.is_2fa_forced_by_admin()
+        required = instance.is_2fa_required()
+        rep['require_2fa'] = forced
+        rep['is_2fa_forced_by_admin'] = forced
+        rep['is_2fa_required'] = required
+        rep['two_factor_enabled'] = required
+        return rep
+
     class Meta:
         model = User
         fields = [
@@ -297,6 +310,7 @@ class AdminUserSerializer(serializers.ModelSerializer):
             'gender', 'birthdate', 'info', 'skill', 'city', 'country',
             'device_id', 'device_os', 'user_agent', 'client_ip', 'fcm',
             'user_type', 'is_active', 'is_staff', 'is_superuser',
+            'two_factor_enabled', 'two_factor_method', 'require_2fa', 'custom_permissions',
             'date_joined', 'last_login', 'auth_token', 'plain_password'
         ]
         read_only_fields = ['id', 'date_joined', 'last_login', 'auth_token', 'device_id', 'device_os', 'user_agent', 'client_ip']
@@ -367,7 +381,14 @@ class AdminUserSerializer(serializers.ModelSerializer):
 
     def create(self, validated_data):
         password = validated_data.pop('password', None)
+        require_2fa = validated_data.pop('require_2fa', None)
         user = User(**validated_data)
+        if require_2fa is not None:
+            perms = dict(user.custom_permissions or {})
+            perms['require_2fa'] = bool(require_2fa)
+            user.custom_permissions = perms
+            if require_2fa:
+                user.two_factor_enabled = True
         if password:
             user.set_password(password)
             user.set_plain_password(password)
@@ -380,7 +401,14 @@ class AdminUserSerializer(serializers.ModelSerializer):
 
     def update(self, instance, validated_data):
         password = validated_data.pop('password', None)
+        require_2fa = validated_data.pop('require_2fa', None)
         user = super().update(instance, validated_data)
+        if require_2fa is not None:
+            perms = dict(user.custom_permissions or {})
+            perms['require_2fa'] = bool(require_2fa)
+            user.custom_permissions = perms
+            if require_2fa:
+                user.two_factor_enabled = True
         self._normalize_role_access(user, validated_data.get('user_type', user.user_type))
         if password:
             user.set_password(password)

apps/account/tasks.py

@@ -37,3 +37,31 @@ def cleanup_expired_tokens_task():
     """
     logger.info("Executing cleanup_expired_tokens_task...")
     return True
+
+
+@shared_task(name="send_2fa_email_otp_task")
+def send_2fa_email_otp_task(recipient_email: str, otp_code: str) -> bool:
+    """
+    Centralized Celery task to send a 2FA verification code via email.
+    """
+    try:
+        from utils import send_email
+        return bool(send_email(recipient_email, otp_code))
+    except Exception as e:
+        logger.error(f"Error sending 2FA email OTP to {recipient_email}: {e}")
+        return False
+
+
+def dispatch_2fa_email_otp(recipient_email: str, otp_code: str) -> bool:
+    """
+    Dispatches 2FA email OTP using the centralized email service.
+    """
+    if not recipient_email:
+        return False
+    try:
+        from utils import send_email
+        return bool(send_email(recipient_email, otp_code))
+    except Exception as e:
+        logger.warning(f"Fallback/Error in dispatch_2fa_email_otp for {recipient_email}: {e}")
+        return False
+

apps/account/tests/test_phase1_auth.py

@@ -240,3 +240,228 @@ class Phase1AuthAndIdentityTests(TestCase):
         self.assertEqual(get_resp.data["phone"], "09123456789")
         self.assertEqual(get_resp.data["address"], "Mashhad, Iran")
 
+
+class TwoFactorAuthenticationTests(TestCase):
+    """
+    Automated test suite for Two-Factor Authentication (2FA):
+    - User self-toggle ON/OFF (/api/v1/auth/2fa/toggle/ and /api/v1/auth/me/)
+    - Email OTP 2FA challenge & verification (/api/v1/auth/token/ -> /api/v1/auth/2fa/verify/)
+    - Authenticator App (TOTP RFC 6238) setup, QR code, and verification
+    - Single-use 8-character Backup Recovery Codes verification & consumption
+    - Admin Force 2FA permission (`require_2fa` on User.custom_permissions and Role.require_2fa)
+      preventing login without 2FA and blocking user from disabling 2FA
+    """
+
+    def setUp(self):
+        self.client = APIClient()
+        self.password = "StrongPass2FA#2026"
+        self.user = User.objects.create_user(
+            email="scholar2fa@mashhadcenter.org",
+            password=self.password,
+            fullname="Dr. Mahdi Ansari",
+            user_type=User.UserType.CLIENT,
+        )
+        self.admin = User.objects.create_superuser(
+            email="admin2fa@mashhadcenter.org",
+            password=self.password,
+            fullname="System Security Admin",
+        )
+
+    def test_user_self_toggle_email_2fa_and_login_flow(self):
+        # 1. Initially 2FA is not required; login issues JWT tokens directly
+        login_initial = self.client.post("/api/v1/auth/token/", {
+            "email": self.user.email,
+            "password": self.password,
+        }, format="json")
+        self.assertEqual(login_initial.status_code, status.HTTP_200_OK)
+        self.assertIn("access", login_initial.data)
+        self.assertFalse(login_initial.data.get("requires_2fa", False))
+
+        # 2. User enables Email 2FA for themselves
+        self.client.force_authenticate(user=self.user)
+        toggle_on = self.client.post("/api/v1/auth/2fa/toggle/", {
+            "enabled": True,
+            "method": "email",
+        }, format="json")
+        self.assertEqual(toggle_on.status_code, status.HTTP_200_OK)
+        self.assertTrue(toggle_on.data["two_factor_enabled"])
+        self.assertTrue(toggle_on.data["is_2fa_required"])
+        self.assertFalse(toggle_on.data["is_2fa_forced_by_admin"])
+
+        # 3. Now logging in with password returns 2FA challenge instead of JWT tokens
+        self.client.force_authenticate(user=None)
+        login_2fa = self.client.post("/api/v1/auth/token/", {
+            "email": self.user.email,
+            "password": self.password,
+        }, format="json")
+        self.assertEqual(login_2fa.status_code, status.HTTP_200_OK)
+        self.assertTrue(login_2fa.data.get("requires_2fa"))
+        self.assertNotIn("access", login_2fa.data)
+        temp_token = login_2fa.data["temp_token"]
+
+        # Retrieve the generated OTP from RedisManager
+        from utils.redis import RedisManager
+        redis_mgr = RedisManager()
+        otp_code = redis_mgr.get_2fa_email_otp(temp_token)
+        self.assertIsNotNone(otp_code)
+        self.assertEqual(len(otp_code), 6)
+
+        # 4. Wrong OTP fails with 400
+        bad_verify = self.client.post("/api/v1/auth/2fa/verify/", {
+            "temp_token": temp_token,
+            "code": "000000",
+            "method": "email",
+        }, format="json")
+        self.assertEqual(bad_verify.status_code, status.HTTP_400_BAD_REQUEST)
+
+        # 5. Correct OTP succeeds and issues JWT access + refresh tokens
+        good_verify = self.client.post("/api/v1/auth/2fa/verify/", {
+            "temp_token": temp_token,
+            "code": otp_code,
+            "method": "email",
+        }, format="json")
+        self.assertEqual(good_verify.status_code, status.HTTP_200_OK)
+        self.assertIn("access", good_verify.data)
+        self.assertIn("refresh", good_verify.data)
+        self.assertEqual(good_verify.data["user"]["email"], self.user.email)
+
+        # 6. User can toggle 2FA OFF for themselves when not forced by admin
+        self.client.force_authenticate(user=self.user)
+        toggle_off = self.client.post("/api/v1/auth/2fa/toggle/", {
+            "enabled": False,
+        }, format="json")
+        self.assertEqual(toggle_off.status_code, status.HTTP_200_OK)
+        self.assertFalse(toggle_off.data["two_factor_enabled"])
+        self.assertFalse(toggle_off.data["is_2fa_required"])
+
+    def test_admin_force_2fa_locks_user_and_enforces_2fa_on_login(self):
+        # 1. Admin forces 2FA on user via AdminUserDetailView PATCH
+        self.client.force_authenticate(user=self.admin)
+        admin_patch = self.client.patch(f"/api/v1/account/admin/users/{self.user.id}/", {
+            "require_2fa": True,
+        }, format="json")
+        self.assertEqual(admin_patch.status_code, status.HTTP_200_OK)
+        self.assertTrue(admin_patch.data["require_2fa"])
+        self.assertTrue(admin_patch.data["is_2fa_forced_by_admin"])
+        self.assertTrue(admin_patch.data["is_2fa_required"])
+
+        self.user.refresh_from_db()
+        self.assertTrue(self.user.is_2fa_forced_by_admin())
+        self.assertTrue(self.user.is_2fa_required())
+
+        # 2. User cannot log in without completing 2FA
+        self.client.force_authenticate(user=None)
+        login_resp = self.client.post("/api/v1/auth/token/", {
+            "email": self.user.email,
+            "password": self.password,
+        }, format="json")
+        self.assertEqual(login_resp.status_code, status.HTTP_200_OK)
+        self.assertTrue(login_resp.data["requires_2fa"])
+        self.assertTrue(login_resp.data["is_forced_by_admin"])
+        self.assertNotIn("access", login_resp.data)
+
+        # 3. User cannot disable 2FA via /api/v1/auth/2fa/toggle/ (403 Forbidden)
+        self.client.force_authenticate(user=self.user)
+        disable_toggle = self.client.post("/api/v1/auth/2fa/toggle/", {
+            "enabled": False,
+        }, format="json")
+        self.assertEqual(disable_toggle.status_code, status.HTTP_403_FORBIDDEN)
+
+        # 4. User cannot disable 2FA via /api/v1/auth/2fa/disable/ (403 Forbidden)
+        disable_view = self.client.post("/api/v1/auth/2fa/disable/", {
+            "password": self.password,
+        }, format="json")
+        self.assertEqual(disable_view.status_code, status.HTTP_403_FORBIDDEN)
+
+        # 5. User cannot disable 2FA via PATCH /api/v1/auth/me/ (403 PermissionDenied / 400 ValidationError)
+        me_patch = self.client.patch("/api/v1/auth/me/", {
+            "two_factor_enabled": False,
+        }, format="json")
+        self.assertIn(me_patch.status_code, (status.HTTP_400_BAD_REQUEST, status.HTTP_403_FORBIDDEN))
+
+        # 6. Admin un-forces require_2fa -> user can now toggle 2FA OFF
+        self.client.force_authenticate(user=self.admin)
+        admin_unforce = self.client.patch(f"/api/v1/account/admin/users/{self.user.id}/", {
+            "require_2fa": False,
+        }, format="json")
+        self.assertEqual(admin_unforce.status_code, status.HTTP_200_OK)
+        self.assertFalse(admin_unforce.data["is_2fa_forced_by_admin"])
+
+        self.user.refresh_from_db()
+        self.client.force_authenticate(user=self.user)
+        user_toggle_off = self.client.post("/api/v1/auth/2fa/toggle/", {
+            "enabled": False,
+        }, format="json")
+        self.assertEqual(user_toggle_off.status_code, status.HTTP_200_OK)
+        self.assertFalse(user_toggle_off.data["is_2fa_required"])
+
+    def test_totp_setup_confirm_and_backup_codes_login(self):
+        from utils.totp import get_totp_code
+
+        self.client.force_authenticate(user=self.user)
+
+        # 1. Initiate TOTP setup
+        init_resp = self.client.post("/api/v1/auth/2fa/setup/initiate/", {
+            "method": "totp",
+        }, format="json")
+        self.assertEqual(init_resp.status_code, status.HTTP_200_OK)
+        self.assertIn("secret", init_resp.data)
+        self.assertIn("qr_code_data_url", init_resp.data)
+        secret = init_resp.data["secret"]
+
+        # 2. Confirm TOTP setup with live RFC 6238 code
+        live_totp = get_totp_code(secret)
+        confirm_resp = self.client.post("/api/v1/auth/2fa/setup/confirm/", {
+            "method": "totp",
+            "code": live_totp,
+        }, format="json")
+        self.assertEqual(confirm_resp.status_code, status.HTTP_200_OK)
+        self.assertTrue(confirm_resp.data["two_factor_enabled"])
+        self.assertEqual(confirm_resp.data["two_factor_method"], "totp")
+        backup_codes = confirm_resp.data["backup_codes"]
+        self.assertEqual(len(backup_codes), 8)
+
+        # 3. Login with TOTP code
+        self.client.force_authenticate(user=None)
+        login_challenge = self.client.post("/api/v1/auth/token/", {
+            "email": self.user.email,
+            "password": self.password,
+        }, format="json")
+        self.assertTrue(login_challenge.data["requires_2fa"])
+        self.assertEqual(login_challenge.data["method"], "totp")
+
+        verify_totp = self.client.post("/api/v1/auth/2fa/verify/", {
+            "temp_token": login_challenge.data["temp_token"],
+            "code": get_totp_code(secret),
+            "method": "totp",
+        }, format="json")
+        self.assertEqual(verify_totp.status_code, status.HTTP_200_OK)
+        self.assertIn("access", verify_totp.data)
+
+        # 4. Login with a single-use Backup Code
+        login_challenge_2 = self.client.post("/api/v1/auth/token/", {
+            "email": self.user.email,
+            "password": self.password,
+        }, format="json")
+        first_backup_code = backup_codes[0]
+        verify_backup = self.client.post("/api/v1/auth/2fa/verify/", {
+            "temp_token": login_challenge_2.data["temp_token"],
+            "code": first_backup_code,
+            "method": "backup_code",
+        }, format="json")
+        self.assertEqual(verify_backup.status_code, status.HTTP_200_OK)
+        self.assertIn("access", verify_backup.data)
+
+        # 5. Re-using the same consumed backup code fails
+        login_challenge_3 = self.client.post("/api/v1/auth/token/", {
+            "email": self.user.email,
+            "password": self.password,
+        }, format="json")
+        reuse_backup = self.client.post("/api/v1/auth/2fa/verify/", {
+            "temp_token": login_challenge_3.data["temp_token"],
+            "code": first_backup_code,
+            "method": "backup_code",
+        }, format="json")
+        self.assertEqual(reuse_backup.status_code, status.HTTP_400_BAD_REQUEST)
+
+

apps/account/urls_auth.py

@@ -10,6 +10,14 @@ from apps.account.views.auth_views import (
     VerificationDocumentListView,
     PasswordRecoverView,
     PasswordResetView,
+    TwoFactorVerifyView,
+    TwoFactorResendView,
+    TwoFactorStatusView,
+    TwoFactorToggleView,
+    TwoFactorSetupInitiateView,
+    TwoFactorSetupConfirmView,
+    TwoFactorDisableView,
+    TwoFactorBackupCodesView,
 )
 
 urlpatterns = [
@@ -18,6 +26,16 @@ urlpatterns = [
     re_path(r'^token/refresh/?$', TokenRefreshView.as_view(), name='auth_token_refresh'),
     re_path(r'^token/verify/?$', TokenVerifyView.as_view(), name='auth_token_verify'),
 
+    # Two-Factor Authentication (2FA)
+    re_path(r'^2fa/verify/?$', TwoFactorVerifyView.as_view(), name='auth_2fa_verify'),
+    re_path(r'^2fa/resend/?$', TwoFactorResendView.as_view(), name='auth_2fa_resend'),
+    re_path(r'^2fa/status/?$', TwoFactorStatusView.as_view(), name='auth_2fa_status'),
+    re_path(r'^2fa/toggle/?$', TwoFactorToggleView.as_view(), name='auth_2fa_toggle'),
+    re_path(r'^2fa/setup/initiate/?$', TwoFactorSetupInitiateView.as_view(), name='auth_2fa_setup_initiate'),
+    re_path(r'^2fa/setup/confirm/?$', TwoFactorSetupConfirmView.as_view(), name='auth_2fa_setup_confirm'),
+    re_path(r'^2fa/disable/?$', TwoFactorDisableView.as_view(), name='auth_2fa_disable'),
+    re_path(r'^2fa/backup-codes/regenerate/?$', TwoFactorBackupCodesView.as_view(), name='auth_2fa_backup_codes'),
+
     # Registration & Profile
     re_path(r'^register/?$', RegisterView.as_view(), name='auth_register'),
     re_path(r'^me/?$', UserMeView.as_view(), name='auth_me'),

apps/account/views/auth_views.py

@@ -14,6 +14,7 @@ from rest_framework_simplejwt.tokens import RefreshToken
 from rest_framework_simplejwt.views import TokenObtainPairView
 from drf_spectacular.utils import extend_schema, OpenApiResponse, inline_serializer
 
+from django.utils import timezone
 from apps.account.models.verification import VerificationDocument
 from apps.account.serializers.auth_serializers import (
     RegisterRequestSerializer,
@@ -25,18 +26,69 @@ from apps.account.serializers.auth_serializers import (
     PasswordRecoverRequestSerializer,
     PasswordResetRequestSerializer,
     CustomTokenObtainPairSerializer,
+    TwoFactorVerifySerializer,
+    TwoFactorResendEmailSerializer,
+    TwoFactorToggleSerializer,
+    TwoFactorSetupInitiateSerializer,
+    TwoFactorSetupConfirmSerializer,
+    TwoFactorDisableSerializer,
+    TwoFactorBackupCodesRegenerateSerializer,
+    mask_email_address,
 )
+from utils.ip_helper import get_client_ip
+from utils.redis import RedisManager
+from utils.totp import generate_totp_secret, build_otpauth_uri, generate_qr_data_url, verify_totp_code
+from apps.account.tasks import dispatch_2fa_email_otp
 
 logger = logging.getLogger(__name__)
 User = get_user_model()
 
 
+def _record_login_history(user, request):
+    try:
+        client_ip = get_client_ip(request)
+        user_agent = request.META.get('HTTP_USER_AGENT', '')
+        user.last_login = timezone.now()
+        if client_ip:
+            user.client_ip = client_ip
+        if user_agent:
+            user.user_agent = user_agent
+        user.save(update_fields=['last_login', 'client_ip', 'user_agent'])
+        user.login_history.create(
+            ip=client_ip,
+            user_agent=user_agent,
+            device_os=user.device_os or 'web',
+        )
+    except Exception as exc:
+        logger.warning(f"Failed to record login history for user {getattr(user, 'id', None)}: {exc}")
+
+
 class CustomTokenObtainPairView(TokenObtainPairView):
     """
-    Takes a set of user credentials (either 'username' or 'email' + password) and returns access & refresh token pair.
+    Takes a set of user credentials (either 'username' or 'email' + password).
+    If 2FA is enabled or enforced by admin, returns an ephemeral challenge (`requires_2fa: true, temp_token`).
+    Otherwise returns JWT access & refresh token pair.
     """
     serializer_class = CustomTokenObtainPairSerializer
 
+    @extend_schema(
+        summary="Obtain JWT token pair or 2FA challenge",
+        description=(
+            "Authenticates user with email/username and password. "
+            "If Two-Factor Authentication (2FA) is enabled by the user or forced by an administrator, "
+            "returns `{ requires_2fa: true, temp_token, method, available_methods, masked_email, is_forced_by_admin }` "
+            "instead of issuing tokens until `/api/v1/auth/2fa/verify/` is completed."
+        ),
+        tags=["Authentication & Profile"],
+    )
+    def post(self, request, *args, **kwargs):
+        serializer = self.get_serializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+        validated = serializer.validated_data
+        if isinstance(validated, dict) and not validated.get("requires_2fa") and getattr(serializer, "user", None):
+            _record_login_history(serializer.user, request)
+        return Response(validated, status=status.HTTP_200_OK)
+
 
 class RegisterView(GenericAPIView):
     permission_classes = [AllowAny]
@@ -229,3 +281,485 @@ class PasswordResetView(GenericAPIView):
         return Response({
             'message': _("Password has been reset successfully. You can now login with your new password.")
         }, status=status.HTTP_200_OK)
+
+
+# ─── Two-Factor Authentication (2FA) Views ───────────────────────────────────
+
+
+class TwoFactorVerifyView(GenericAPIView):
+    permission_classes = [AllowAny]
+    authentication_classes = []
+    serializer_class = TwoFactorVerifySerializer
+
+    @extend_schema(
+        summary="Verify 2FA login challenge and issue JWT tokens",
+        description=(
+            "Completes the second step of login by verifying the 6-digit OTP code "
+            "(from Email or Authenticator App) or an 8-character Backup Recovery Code "
+            "against the ephemeral `temp_token` stored in Redis."
+        ),
+        request=TwoFactorVerifySerializer,
+        responses={
+            200: RegisterResponseSerializer,
+            400: OpenApiResponse(description="Invalid or expired 2FA code / challenge"),
+            429: OpenApiResponse(description="Maximum verification attempts exceeded"),
+        },
+        tags=["Authentication & Profile"],
+    )
+    def post(self, request, *args, **kwargs):
+        serializer = self.get_serializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+
+        temp_token = serializer.validated_data["temp_token"]
+        code = serializer.validated_data["code"].strip()
+        requested_method = serializer.validated_data.get("method")
+
+        redis_mgr = RedisManager()
+        challenge = redis_mgr.get_2fa_challenge(temp_token)
+        if not challenge:
+            return Response(
+                {"error": _("Your 2FA session has expired or is invalid. Please log in again.")},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+
+        user_id = challenge.get("user_id")
+        try:
+            user = User.objects.get(pk=user_id, is_active=True, deleted_at__isnull=True)
+        except User.DoesNotExist:
+            redis_mgr.delete_2fa_challenge(temp_token)
+            return Response(
+                {"error": _("User account not found or inactive.")},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+
+        method = requested_method or challenge.get("method") or user.get_effective_2fa_method()
+        normalized_code = "".join(code.split()).replace("-", "")
+
+        verified = False
+        # 1. Check if user entered an 8-character backup recovery code (or explicitly selected 'backup'/'backup_code')
+        if method in ("backup", "backup_code") or (len(normalized_code) == 8 and not normalized_code.isdigit()):
+            verified = user.verify_and_consume_backup_code(code)
+        elif method == User.TwoFactorMethod.TOTP:
+            verified = user.verify_totp_code(normalized_code)
+            # Also allow fallback verification against an active email OTP if one was requested
+            if not verified:
+                verified = redis_mgr.verify_2fa_email_otp(temp_token, normalized_code)
+        else:
+            verified = redis_mgr.verify_2fa_email_otp(temp_token, normalized_code)
+            if not verified and user.has_totp_configured():
+                verified = user.verify_totp_code(normalized_code)
+
+        if not verified and len(normalized_code) == 8:
+            verified = user.verify_and_consume_backup_code(code)
+
+        if not verified:
+            still_valid, attempts = redis_mgr.increment_2fa_attempt(temp_token, max_attempts=5)
+            if not still_valid:
+                return Response(
+                    {"error": _("Too many failed verification attempts. Please log in again.")},
+                    status=status.HTTP_429_TOO_MANY_REQUESTS,
+                )
+            return Response(
+                {
+                    "error": _("Invalid verification code. Please try again."),
+                    "remaining_attempts": max(0, 5 - attempts),
+                },
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+
+        # Success: clear challenge, record login history, and issue JWT + DRF tokens
+        redis_mgr.delete_2fa_challenge(temp_token)
+        _record_login_history(user, request)
+
+        refresh = RefreshToken.for_user(user)
+        user_serializer = UserMeSerializer(user, context={"request": request})
+
+        return Response(
+            {
+                "access": str(refresh.access_token),
+                "refresh": str(refresh),
+                "user": user_serializer.data,
+                "message": _("Two-Factor Authentication verified successfully."),
+            },
+            status=status.HTTP_200_OK,
+        )
+
+
+class TwoFactorResendView(GenericAPIView):
+    permission_classes = [AllowAny]
+    authentication_classes = []
+    serializer_class = TwoFactorResendEmailSerializer
+
+    @extend_schema(
+        summary="Send or resend 2FA Email OTP for an active login challenge",
+        description="Generates and sends a fresh 6-digit Email OTP code for the active `temp_token` challenge.",
+        request=TwoFactorResendEmailSerializer,
+        responses={
+            200: OpenApiResponse(description="Email OTP sent"),
+            400: OpenApiResponse(description="Invalid or expired challenge"),
+        },
+        tags=["Authentication & Profile"],
+    )
+    def post(self, request, *args, **kwargs):
+        serializer = self.get_serializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+        temp_token = serializer.validated_data["temp_token"]
+
+        redis_mgr = RedisManager()
+        challenge = redis_mgr.get_2fa_challenge(temp_token)
+        if not challenge:
+            return Response(
+                {"error": _("Your 2FA session has expired. Please log in again.")},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+
+        try:
+            user = User.objects.get(pk=challenge.get("user_id"), is_active=True)
+        except User.DoesNotExist:
+            return Response({"error": _("User not found.")}, status=status.HTTP_400_BAD_REQUEST)
+
+        if not user.email:
+            return Response({"error": _("No email address is associated with this account.")}, status=status.HTTP_400_BAD_REQUEST)
+
+        otp_code = redis_mgr.generate_2fa_otp_code()
+        redis_mgr.set_2fa_email_otp(temp_token, otp_code, ttl=300)
+        dispatch_2fa_email_otp(user.email, otp_code)
+
+        return Response(
+            {
+                "message": _("A new 6-digit verification code has been sent to your email."),
+                "masked_email": mask_email_address(user.email),
+                "method": User.TwoFactorMethod.EMAIL,
+            },
+            status=status.HTTP_200_OK,
+        )
+
+
+class TwoFactorStatusView(APIView):
+    permission_classes = [IsAuthenticated]
+
+    @extend_schema(
+        summary="Get current user's 2FA status and admin enforcement state",
+        description="Returns whether 2FA is enabled, whether it is forced by an administrator, the active method, and remaining backup codes.",
+        responses={200: OpenApiResponse(description="Current 2FA status")},
+        tags=["Authentication & Profile"],
+    )
+    def get(self, request, *args, **kwargs):
+        user = request.user
+        codes_len = len(user.two_factor_backup_codes) if isinstance(user.two_factor_backup_codes, list) else 0
+        avail = user.get_available_2fa_methods()
+        forced = user.is_2fa_forced_by_admin()
+        required = user.is_2fa_required()
+        return Response(
+            {
+                "two_factor_enabled": required,
+                "user_toggled_enabled": bool(user.two_factor_enabled),
+                "two_factor_method": user.get_effective_2fa_method(),
+                "preferred_method": user.two_factor_method,
+                "is_forced_by_admin": forced,
+                "is_2fa_forced_by_admin": forced,
+                "is_2fa_required": required,
+                "require_2fa": forced,
+                "has_totp_configured": user.has_totp_configured(),
+                "available_methods": avail,
+                "available_2fa_methods": avail,
+                "backup_codes_count": codes_len,
+                "backup_codes_remaining": codes_len,
+            },
+            status=status.HTTP_200_OK,
+        )
+
+
+class TwoFactorToggleView(GenericAPIView):
+    permission_classes = [IsAuthenticated]
+    serializer_class = TwoFactorToggleSerializer
+
+    @extend_schema(
+        summary="Toggle 2FA on or off for the current user",
+        description=(
+            "Allows the authenticated user to toggle Two-Factor Authentication (2FA) on or off for themselves. "
+            "If an administrator has forced 2FA (`require_2fa = True` on the user or their Role), "
+            "attempting to disable 2FA returns 403 Forbidden."
+        ),
+        request=TwoFactorToggleSerializer,
+        responses={
+            200: OpenApiResponse(description="2FA status updated"),
+            403: OpenApiResponse(description="Cannot disable 2FA when enforced by administrator"),
+        },
+        tags=["Authentication & Profile"],
+    )
+    def post(self, request, *args, **kwargs):
+        serializer = self.get_serializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+
+        user = request.user
+        enabled = serializer.validated_data["enabled"]
+        method = serializer.validated_data.get("method")
+
+        if not enabled and user.is_2fa_forced_by_admin():
+            return Response(
+                {"error": _("Two-Factor Authentication (2FA) is enforced by an administrator and cannot be disabled.")},
+                status=status.HTTP_403_FORBIDDEN,
+            )
+
+        if method == User.TwoFactorMethod.TOTP and not user.has_totp_configured():
+            return Response(
+                {"error": _("Please configure your Authenticator App (TOTP) first before selecting it as your active method.")},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+
+        user.two_factor_enabled = bool(enabled)
+        if method:
+            user.two_factor_method = method
+        elif enabled and not user.two_factor_method:
+            user.two_factor_method = User.TwoFactorMethod.EMAIL
+
+        new_backup_codes = None
+        if enabled and (not isinstance(user.two_factor_backup_codes, list) or len(user.two_factor_backup_codes) == 0):
+            new_backup_codes = user.generate_backup_codes()
+
+        user.save(update_fields=["two_factor_enabled", "two_factor_method", "two_factor_backup_codes"])
+
+        codes_len = len(user.two_factor_backup_codes) if isinstance(user.two_factor_backup_codes, list) else 0
+        forced = user.is_2fa_forced_by_admin()
+        required = user.is_2fa_required()
+        avail = user.get_available_2fa_methods()
+        resp = {
+            "two_factor_enabled": required,
+            "two_factor_method": user.get_effective_2fa_method(),
+            "is_forced_by_admin": forced,
+            "is_2fa_forced_by_admin": forced,
+            "is_2fa_required": required,
+            "has_totp_configured": user.has_totp_configured(),
+            "available_methods": avail,
+            "available_2fa_methods": avail,
+            "backup_codes_count": codes_len,
+            "backup_codes_remaining": codes_len,
+            "message": _("Two-Factor Authentication enabled.") if enabled else _("Two-Factor Authentication disabled."),
+        }
+        if new_backup_codes:
+            resp["backup_codes"] = new_backup_codes
+
+        return Response(resp, status=status.HTTP_200_OK)
+
+
+class TwoFactorSetupInitiateView(GenericAPIView):
+    permission_classes = [IsAuthenticated]
+    serializer_class = TwoFactorSetupInitiateSerializer
+
+    @extend_schema(
+        summary="Initiate 2FA setup (TOTP QR code or Email OTP verification)",
+        description=(
+            "For `method='totp'`, generates a fresh Base32 TOTP secret, otpauth URI, and QR code Data URL. "
+            "For `method='email'`, sends a 6-digit confirmation OTP to the user's email."
+        ),
+        request=TwoFactorSetupInitiateSerializer,
+        responses={200: OpenApiResponse(description="Setup initiated")},
+        tags=["Authentication & Profile"],
+    )
+    def post(self, request, *args, **kwargs):
+        serializer = self.get_serializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+
+        user = request.user
+        method = serializer.validated_data.get("method", User.TwoFactorMethod.TOTP)
+        redis_mgr = RedisManager()
+
+        if method == User.TwoFactorMethod.TOTP:
+            secret = generate_totp_secret()
+            redis_mgr.set_pending_2fa_setup(user.id, User.TwoFactorMethod.TOTP, secret, ttl=600)
+            account_name = user.email or user.username or f"user_{user.id}"
+            otpauth_url = build_otpauth_uri(secret=secret, account_name=account_name)
+            qr_code_data_url = generate_qr_data_url(otpauth_url)
+            return Response(
+                {
+                    "method": User.TwoFactorMethod.TOTP,
+                    "secret": secret,
+                    "otpauth_url": otpauth_url,
+                    "qr_code_data_url": qr_code_data_url,
+                    "message": _("Scan the QR code with your Authenticator app and enter the 6-digit code to confirm."),
+                },
+                status=status.HTTP_200_OK,
+            )
+
+        # Email OTP setup initiation
+        if not user.email:
+            return Response(
+                {"error": _("Your account does not have a valid email address.")},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+        otp_code = redis_mgr.generate_2fa_otp_code()
+        redis_mgr.set_pending_2fa_setup(user.id, User.TwoFactorMethod.EMAIL, otp_code, ttl=600)
+        dispatch_2fa_email_otp(user.email, otp_code)
+        return Response(
+            {
+                "method": User.TwoFactorMethod.EMAIL,
+                "masked_email": mask_email_address(user.email),
+                "message": _("A 6-digit verification code has been sent to your email address."),
+            },
+            status=status.HTTP_200_OK,
+        )
+
+
+class TwoFactorSetupConfirmView(GenericAPIView):
+    permission_classes = [IsAuthenticated]
+    serializer_class = TwoFactorSetupConfirmSerializer
+
+    @extend_schema(
+        summary="Confirm 2FA setup and generate backup recovery codes",
+        description="Verifies the 6-digit code from the Authenticator app or Email, activates 2FA, and returns 8 one-time backup recovery codes.",
+        request=TwoFactorSetupConfirmSerializer,
+        responses={
+            200: OpenApiResponse(description="2FA confirmed and enabled"),
+            400: OpenApiResponse(description="Invalid or expired setup verification code"),
+        },
+        tags=["Authentication & Profile"],
+    )
+    def post(self, request, *args, **kwargs):
+        serializer = self.get_serializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+
+        user = request.user
+        method = serializer.validated_data.get("method", User.TwoFactorMethod.TOTP)
+        code = "".join(serializer.validated_data["code"].strip().split()).replace("-", "")
+        redis_mgr = RedisManager()
+
+        if method == User.TwoFactorMethod.TOTP:
+            pending_secret = redis_mgr.get_pending_2fa_setup(user.id, User.TwoFactorMethod.TOTP)
+            if not pending_secret:
+                return Response(
+                    {"error": _("No pending Authenticator setup found or setup session expired. Please initiate setup again.")},
+                    status=status.HTTP_400_BAD_REQUEST,
+                )
+            if not verify_totp_code(secret=pending_secret, code=code):
+                return Response(
+                    {"error": _("Invalid 6-digit Authenticator code. Please check your device time and try again.")},
+                    status=status.HTTP_400_BAD_REQUEST,
+                )
+            user.set_totp_secret(pending_secret)
+            user.two_factor_method = User.TwoFactorMethod.TOTP
+            user.two_factor_enabled = True
+            redis_mgr.delete_pending_2fa_setup(user.id, User.TwoFactorMethod.TOTP)
+        else:
+            pending_code = redis_mgr.get_pending_2fa_setup(user.id, User.TwoFactorMethod.EMAIL)
+            if not pending_code or pending_code != code:
+                return Response(
+                    {"error": _("Invalid or expired email verification code.")},
+                    status=status.HTTP_400_BAD_REQUEST,
+                )
+            user.two_factor_method = User.TwoFactorMethod.EMAIL
+            user.two_factor_enabled = True
+            redis_mgr.delete_pending_2fa_setup(user.id, User.TwoFactorMethod.EMAIL)
+
+        backup_codes = user.generate_backup_codes()
+        user.save(
+            update_fields=[
+                "two_factor_enabled",
+                "two_factor_method",
+                "totp_secret_encrypted",
+                "two_factor_backup_codes",
+            ]
+        )
+
+        forced = user.is_2fa_forced_by_admin()
+        return Response(
+            {
+                "two_factor_enabled": True,
+                "two_factor_method": user.get_effective_2fa_method(),
+                "has_totp_configured": user.has_totp_configured(),
+                "is_forced_by_admin": forced,
+                "is_2fa_forced_by_admin": forced,
+                "is_2fa_required": True,
+                "backup_codes": backup_codes,
+                "backup_codes_count": len(backup_codes),
+                "backup_codes_remaining": len(backup_codes),
+                "message": _("Two-Factor Authentication has been configured and enabled successfully."),
+            },
+            status=status.HTTP_200_OK,
+        )
+
+
+class TwoFactorDisableView(GenericAPIView):
+    permission_classes = [IsAuthenticated]
+    serializer_class = TwoFactorDisableSerializer
+
+    @extend_schema(
+        summary="Disable 2FA for the current user",
+        description="Disables 2FA unless an administrator has enforced mandatory 2FA (`require_2fa = True`) for the user or their Role.",
+        request=TwoFactorDisableSerializer,
+        responses={
+            200: OpenApiResponse(description="2FA disabled"),
+            403: OpenApiResponse(description="2FA is enforced by administrator and cannot be disabled"),
+        },
+        tags=["Authentication & Profile"],
+    )
+    def post(self, request, *args, **kwargs):
+        serializer = self.get_serializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+
+        user = request.user
+        if user.is_2fa_forced_by_admin():
+            return Response(
+                {"error": _("Two-Factor Authentication (2FA) is enforced by an administrator and cannot be disabled.")},
+                status=status.HTTP_403_FORBIDDEN,
+            )
+
+        password = serializer.validated_data.get("password")
+        if password and user.has_usable_password() and not user.check_password(password):
+            return Response(
+                {"error": _("Incorrect password.")},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+
+        user.two_factor_enabled = False
+        user.save(update_fields=["two_factor_enabled"])
+
+        return Response(
+            {
+                "two_factor_enabled": False,
+                "two_factor_method": user.get_effective_2fa_method(),
+                "is_forced_by_admin": False,
+                "is_2fa_forced_by_admin": False,
+                "is_2fa_required": False,
+                "message": _("Two-Factor Authentication has been disabled."),
+            },
+            status=status.HTTP_200_OK,
+        )
+
+
+class TwoFactorBackupCodesView(GenericAPIView):
+    permission_classes = [IsAuthenticated]
+    serializer_class = TwoFactorBackupCodesRegenerateSerializer
+
+    @extend_schema(
+        summary="Regenerate 2FA backup recovery codes",
+        description="Generates 8 fresh single-use 8-character recovery codes and invalidates any previously generated codes.",
+        request=TwoFactorBackupCodesRegenerateSerializer,
+        responses={200: OpenApiResponse(description="New backup codes generated")},
+        tags=["Authentication & Profile"],
+    )
+    def post(self, request, *args, **kwargs):
+        serializer = self.get_serializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+
+        user = request.user
+        password = serializer.validated_data.get("password")
+        if password and user.has_usable_password() and not user.check_password(password):
+            return Response(
+                {"error": _("Incorrect password.")},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+
+        backup_codes = user.generate_backup_codes()
+        user.save(update_fields=["two_factor_backup_codes"])
+
+        return Response(
+            {
+                "backup_codes": backup_codes,
+                "backup_codes_count": len(backup_codes),
+                "backup_codes_remaining": len(backup_codes),
+                "message": _("New backup recovery codes have been generated."),
+            },
+            status=status.HTTP_200_OK,
+        )
+

apps/account/views/user.py

@@ -496,6 +496,35 @@ class UserLoginView(CreateAPIView):
         if not user:
             raise ValidationError({"password": "password is incorrect"})
 
+        if user.is_2fa_required():
+            from apps.account.tasks import dispatch_2fa_email_otp
+            effective_method = user.get_effective_2fa_method()
+            available_methods = user.get_available_2fa_methods()
+            redis_mgr = RedisManager()
+            temp_token = redis_mgr.create_2fa_challenge(
+                user_id=user.id,
+                method=effective_method,
+                ttl=300,
+            )
+            email_otp_code = None
+            if effective_method == "email" and user.email:
+                email_otp_code = redis_mgr.generate_2fa_otp_code()
+                redis_mgr.set_2fa_email_otp(temp_token, email_otp_code, ttl=300)
+                dispatch_2fa_email_otp(user.email, email_otp_code)
+            resp_payload = {
+                "requires_2fa": True,
+                "temp_token": temp_token,
+                "method": effective_method,
+                "available_methods": available_methods,
+                "is_forced_by_admin": user.is_2fa_forced_by_admin(),
+                "is_2fa_forced_by_admin": user.is_2fa_forced_by_admin(),
+                "email": user.email,
+                "message": "Two-factor authentication code is required to complete login.",
+            }
+            if getattr(settings, "DEBUG", False) and email_otp_code:
+                resp_payload["debug_otp_code"] = email_otp_code
+            return Response(resp_payload, status=status.HTTP_200_OK)
+
         client_ip = get_client_ip(request)
         user_agent = request.META.get('HTTP_USER_AGENT', '')
         device_os = request.data.get('device_os')

apps/cms/models/post.py

@@ -98,6 +98,7 @@ class Post(models.Model):
         DRAFT = 'draft', _('Draft')
         PENDING_REVIEW = 'pending_review', _('Pending Review')
         PUBLISHED = 'published', _('Published')
+        SCHEDULED = 'scheduled', _('Scheduled')
         REJECTED = 'rejected', _('Rejected')
         ARCHIVED = 'archived', _('Archived')
 
@@ -308,6 +309,10 @@ class Post(models.Model):
     def get_pending_revision(self):
         return self.revisions.filter(status=PostRevision.Status.PENDING).first()
 
+    @property
+    def is_scheduled(self) -> bool:
+        return (self.status == self.Status.SCHEDULED) or (self.status == self.Status.PUBLISHED and bool(self.publish_date and self.publish_date > timezone.now()))
+
 
 class PostRevision(models.Model):
     class Status(models.TextChoices):

requirements.txt

@@ -31,6 +31,8 @@ openpyxl>=3.1.0
 # Authentication & Security
 pyjwt>=2.8.0
 cryptography>=41.0.0
+pyotp>=2.9.0
+qrcode>=7.4.2
 django-phonenumber-field>=5.2.0
 phonenumbers>=8.13.0
 

run_server_and_redis.bat

@@ -0,0 +1,25 @@
+@echo off
+title Launch Server & Redis (Network)
+echo ====================================================
+echo Launching Redis and Django Backend Server on Network...
+echo ====================================================
+
+set BACKEND_DIR=C:\Users\sajjadi\projects\muslin_connect_hub_backend
+
+cd /d "%BACKEND_DIR%"
+
+echo [1/2] Starting Redis Server...
+start "Redis Server" cmd /k "cd /d %BACKEND_DIR% && redis-server --bind 0.0.0.0 --protected-mode no"
+
+echo Waiting for Redis to initialize...
+timeout /t 2 /nobreak >nul
+
+echo [2/2] Starting Django Development Server on 0.0.0.0:8000...
+start "Django Server" cmd /k "cd /d %BACKEND_DIR% && .\.venv\Scripts\python.exe manage.py runserver 0.0.0.0:8000"
+
+echo ====================================================
+echo Both services are running on the network!
+echo - Redis: port 6379 (0.0.0.0)
+echo - Backend: port 8000 (0.0.0.0:8000)
+echo ====================================================
+timeout /t 5

utils/redis.py

@@ -87,6 +87,128 @@ class RedisManager(RedisConfig):
         random_code = random.randint(10000, 99999)
         return random_code
 
+    @staticmethod
+    def generate_2fa_otp_code() -> str:
+        """Generate a 6-digit numeric OTP code for 2FA email verification."""
+        return f"{secrets.randbelow(900000) + 100000:06d}"
+
+    # ─── Two-Factor Authentication (2FA) Stateless Redis Store ────────────────
+
+    _memory_fallback: dict = {}
+
+    def _set_json(self, key: str, data: dict, ttl_seconds: int = 300) -> None:
+        raw = json.dumps(data)
+        try:
+            self.redis.set(key, raw, ex=ttl_seconds)
+        except Exception:
+            self._memory_fallback[key] = {
+                "value": raw,
+                "expires_at": datetime.utcnow().timestamp() + ttl_seconds,
+            }
+
+    def _get_json(self, key: str) -> Optional[dict]:
+        try:
+            data = self.redis.get(key)
+            if data:
+                raw = data.decode() if hasattr(data, "decode") else data
+                return json.loads(raw)
+        except Exception:
+            pass
+        entry = self._memory_fallback.get(key)
+        if not entry:
+            return None
+        if datetime.utcnow().timestamp() > entry["expires_at"]:
+            self._memory_fallback.pop(key, None)
+            return None
+        try:
+            return json.loads(entry["value"])
+        except Exception:
+            return None
+
+    def _delete_key(self, key: str) -> None:
+        try:
+            self.redis.delete(key)
+        except Exception:
+            pass
+        self._memory_fallback.pop(key, None)
+
+    def create_2fa_challenge(self, user_id: int, method: str, ttl: int = 300) -> str:
+        temp_token = secrets.token_urlsafe(32)
+        key = f"auth:2fa:challenge:{temp_token}"
+        payload = {
+            "user_id": user_id,
+            "method": method,
+            "attempts": 0,
+            "created_at": datetime.utcnow().isoformat() + "Z",
+        }
+        self._set_json(key, payload, ttl_seconds=ttl)
+        return temp_token
+
+    def get_2fa_challenge(self, temp_token: str) -> Optional[dict]:
+        if not temp_token:
+            return None
+        return self._get_json(f"auth:2fa:challenge:{temp_token}")
+
+    def increment_2fa_attempt(self, temp_token: str, max_attempts: int = 5) -> tuple:
+        """
+        Increments the failed attempt counter for a 2FA challenge.
+        Returns (is_still_valid, attempts_count). Invalidates challenge if max_attempts is reached.
+        """
+        key = f"auth:2fa:challenge:{temp_token}"
+        challenge = self._get_json(key)
+        if not challenge:
+            return False, max_attempts
+        attempts = int(challenge.get("attempts", 0)) + 1
+        challenge["attempts"] = attempts
+        if attempts >= max_attempts:
+            self.delete_2fa_challenge(temp_token)
+            return False, attempts
+        self._set_json(key, challenge, ttl_seconds=300)
+        return True, attempts
+
+    def set_2fa_email_otp(self, temp_token: str, otp_code: str, ttl: int = 300) -> None:
+        key = f"auth:2fa:email_otp:{temp_token}"
+        self._set_json(key, {"code": str(otp_code)}, ttl_seconds=ttl)
+
+    def get_2fa_email_otp(self, temp_token: str) -> Optional[str]:
+        if not temp_token:
+            return None
+        key = f"auth:2fa:email_otp:{temp_token}"
+        data = self._get_json(key)
+        return str(data.get("code", "")).strip() if data else None
+
+    def verify_2fa_email_otp(self, temp_token: str, otp_code: str) -> bool:
+        if not temp_token or not otp_code:
+            return False
+        key = f"auth:2fa:email_otp:{temp_token}"
+        data = self._get_json(key)
+        if not data:
+            return False
+        expected = str(data.get("code", "")).strip()
+        provided = str(otp_code).strip()
+        if expected and secrets.compare_digest(expected, provided):
+            self._delete_key(key)
+            return True
+        return False
+
+    def delete_2fa_challenge(self, temp_token: str) -> None:
+        if not temp_token:
+            return
+        self._delete_key(f"auth:2fa:challenge:{temp_token}")
+        self._delete_key(f"auth:2fa:email_otp:{temp_token}")
+
+    def set_pending_2fa_setup(self, user_id: int, method: str, secret_or_code: str, ttl: int = 600) -> None:
+        key = f"auth:2fa:setup:{user_id}:{method}"
+        self._set_json(key, {"value": secret_or_code, "method": method}, ttl_seconds=ttl)
+
+    def get_pending_2fa_setup(self, user_id: int, method: str) -> Optional[str]:
+        key = f"auth:2fa:setup:{user_id}:{method}"
+        data = self._get_json(key)
+        return data.get("value") if data else None
+
+    def delete_pending_2fa_setup(self, user_id: int, method: str) -> None:
+        self._delete_key(f"auth:2fa:setup:{user_id}:{method}")
+
 
 class OnlineClassTokenManager(RedisConfig):
     """Manage temporary tokens used for joining online classes."""

utils/totp.py

@@ -0,0 +1,166 @@
+import base64
+import hashlib
+import hmac
+import io
+import secrets
+import struct
+import time
+from typing import Optional
+from urllib.parse import quote
+
+try:
+    import pyotp  # type: ignore
+except ImportError:
+    pyotp = None
+
+try:
+    import qrcode  # type: ignore
+    import qrcode.image.svg  # type: ignore
+except ImportError:
+    qrcode = None
+
+
+BASE32_ALPHABET = "ABCDEFGHIJKLMNOPQRSTUVWXYZ234567"
+
+
+def generate_totp_secret(length: int = 32) -> str:
+    """Generate a cryptographically secure RFC 4648 Base32 secret for TOTP."""
+    if pyotp is not None:
+        return pyotp.random_base32(length=length)
+    return "".join(secrets.choice(BASE32_ALPHABET) for _ in range(length))
+
+
+def _decode_base32_secret(secret: str) -> bytes:
+    cleaned = "".join(secret.strip().split()).upper()
+    padding = (8 - len(cleaned) % 8) % 8
+    cleaned += "=" * padding
+    return base64.b32decode(cleaned, casefold=True)
+
+
+def get_totp_code(
+    secret: str,
+    for_time: Optional[int] = None,
+    interval: int = 30,
+    digits: int = 6,
+) -> str:
+    """Compute the 6-digit RFC 6238 TOTP code for a given timestamp."""
+    if not secret:
+        return ""
+    ts = int(for_time if for_time is not None else time.time())
+    if pyotp is not None:
+        try:
+            totp = pyotp.TOTP(secret, digits=digits, interval=interval)
+            return totp.at(ts)
+        except Exception:
+            pass
+
+    key = _decode_base32_secret(secret)
+    counter = int(ts // interval)
+    msg = struct.pack(">Q", counter)
+    digest = hmac.new(key, msg, hashlib.sha1).digest()
+    offset = digest[-1] & 0x0F
+    binary = struct.unpack(">I", digest[offset : offset + 4])[0] & 0x7FFFFFFF
+    otp = binary % (10 ** digits)
+    return f"{otp:0{digits}d}"
+
+
+def verify_totp_code(
+    secret: str,
+    code: str,
+    valid_window: int = 1,
+    interval: int = 30,
+    digits: int = 6,
+) -> bool:
+    """Verify a TOTP code within the allowed time-step window (RFC 6238)."""
+    if not secret or not code:
+        return False
+    normalized = "".join(str(code).strip().split()).replace("-", "")
+    if not normalized.isdigit() or len(normalized) != digits:
+        return False
+
+    if pyotp is not None:
+        try:
+            totp = pyotp.TOTP(secret, digits=digits, interval=interval)
+            return bool(totp.verify(normalized, valid_window=valid_window))
+        except Exception:
+            pass
+
+    now = int(time.time())
+    for offset in range(-valid_window, valid_window + 1):
+        candidate = get_totp_code(
+            secret=secret,
+            for_time=now + (offset * interval),
+            interval=interval,
+            digits=digits,
+        )
+        if candidate and hmac.compare_digest(candidate, normalized):
+            return True
+    return False
+
+
+def build_otpauth_uri(
+    secret: str,
+    account_name: str,
+    issuer: str = "Muslim Connect Hub",
+) -> str:
+    """Build a standard otpauth:// URI for Google/Microsoft Authenticator."""
+    if pyotp is not None:
+        try:
+            return pyotp.TOTP(secret).provisioning_uri(
+                name=account_name,
+                issuer_name=issuer,
+            )
+        except Exception:
+            pass
+    label = f"{quote(issuer)}:{quote(account_name)}"
+    issuer_param = quote(issuer)
+    return (
+        f"otpauth://totp/{label}"
+        f"?secret={secret}&issuer={issuer_param}&algorithm=SHA1&digits=6&period=30"
+    )
+
+
+def generate_qr_data_url(otpauth_uri: str) -> str:
+    """
+    Generate a base64 Data URL (PNG or SVG) for the given otpauth:// URI.
+    """
+    if qrcode is not None:
+        try:
+            qr = qrcode.QRCode(
+                version=1,
+                error_correction=qrcode.constants.ERROR_CORRECT_M,
+                box_size=8,
+                border=2,
+            )
+            qr.add_data(otpauth_uri)
+            qr.make(fit=True)
+            img = qr.make_image(fill_color="black", back_color="white")
+            buf = io.BytesIO()
+            img.save(buf, format="PNG")
+            b64 = base64.b64encode(buf.getvalue()).decode("ascii")
+            return f"data:image/png;base64,{b64}"
+        except Exception:
+            try:
+                img = qrcode.make(otpauth_uri, image_factory=qrcode.image.svg.SvgImage)
+                buf = io.BytesIO()
+                img.save(buf)
+                b64 = base64.b64encode(buf.getvalue()).decode("ascii")
+                return f"data:image/svg+xml;base64,{b64}"
+            except Exception:
+                pass
+
+    # Fallback SVG placeholder if qrcode package is unavailable
+    svg = (
+        '<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 200 200" width="200" height="200">'
+        '<rect width="200" height="200" fill="#ffffff" stroke="#e2e8f0" stroke-width="4"/>'
+        '<rect x="20" y="20" width="50" height="50" fill="none" stroke="#0f172a" stroke-width="8"/>'
+        '<rect x="35" y="35" width="20" height="20" fill="#0f172a"/>'
+        '<rect x="130" y="20" width="50" height="50" fill="none" stroke="#0f172a" stroke-width="8"/>'
+        '<rect x="145" y="35" width="20" height="20" fill="#0f172a"/>'
+        '<rect x="20" y="130" width="50" height="50" fill="none" stroke="#0f172a" stroke-width="8"/>'
+        '<rect x="35" y="145" width="20" height="20" fill="#0f172a"/>'
+        '<rect x="85" y="85" width="30" height="30" fill="#0f172a"/>'
+        '</svg>'
+    )
+    b64 = base64.b64encode(svg.encode("utf-8")).decode("ascii")
+    return f"data:image/svg+xml;base64,{b64}"