diff --git a/apps/account/admin/role.py b/apps/account/admin/role.py index aa8036a..094843e 100644 --- a/apps/account/admin/role.py +++ b/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"), { diff --git a/apps/account/admin/user.py b/apps/account/admin/user.py index 210eaf9..5c17ac0 100644 --- a/apps/account/admin/user.py +++ b/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", diff --git a/apps/account/management/commands/seed_roles.py b/apps/account/management/commands/seed_roles.py index 2e52250..0dafa4d 100644 --- a/apps/account/management/commands/seed_roles.py +++ b/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} }, ] diff --git a/apps/account/migrations/0007_role_require_2fa_and_user_2fa_fields.py b/apps/account/migrations/0007_role_require_2fa_and_user_2fa_fields.py new file mode 100644 index 0000000..7981d2e --- /dev/null +++ b/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)', + ), + ), + ] diff --git a/apps/account/models/role.py b/apps/account/models/role.py index e508bbf..2f1c963 100644 --- a/apps/account/models/role.py +++ b/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")) diff --git a/apps/account/models/user.py b/apps/account/models/user.py index 9c02ba6..2a32296 100644 --- a/apps/account/models/user.py +++ b/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",) diff --git a/apps/account/serializers/auth_serializers.py b/apps/account/serializers/auth_serializers.py index ea3080e..eaaf4d9 100644 --- a/apps/account/serializers/auth_serializers.py +++ b/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.") + diff --git a/apps/account/serializers/user.py b/apps/account/serializers/user.py index d378749..816827a 100644 --- a/apps/account/serializers/user.py +++ b/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) diff --git a/apps/account/tasks.py b/apps/account/tasks.py index da54cb6..af70fb6 100644 --- a/apps/account/tasks.py +++ b/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 + diff --git a/apps/account/tests/test_phase1_auth.py b/apps/account/tests/test_phase1_auth.py index e8e3e03..ee1c0a7 100644 --- a/apps/account/tests/test_phase1_auth.py +++ b/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) + + diff --git a/apps/account/urls_auth.py b/apps/account/urls_auth.py index a63f4c7..8528e23 100644 --- a/apps/account/urls_auth.py +++ b/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'), diff --git a/apps/account/views/auth_views.py b/apps/account/views/auth_views.py index 868aee7..283eab3 100644 --- a/apps/account/views/auth_views.py +++ b/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, + ) + diff --git a/apps/account/views/user.py b/apps/account/views/user.py index bc9c0d5..032c7fd 100644 --- a/apps/account/views/user.py +++ b/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') diff --git a/apps/cms/models/post.py b/apps/cms/models/post.py index cda30c5..1566476 100644 --- a/apps/cms/models/post.py +++ b/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): diff --git a/dump.rdb b/dump.rdb index 4f0840d..29f7625 100644 Binary files a/dump.rdb and b/dump.rdb differ diff --git a/requirements.txt b/requirements.txt index 99816a0..209886c 100644 --- a/requirements.txt +++ b/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 diff --git a/run_server_and_redis.bat b/run_server_and_redis.bat new file mode 100644 index 0000000..0594ae8 --- /dev/null +++ b/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 diff --git a/utils/redis.py b/utils/redis.py index cc9fc04..40e2250 100644 --- a/utils/redis.py +++ b/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.""" diff --git a/utils/totp.py b/utils/totp.py new file mode 100644 index 0000000..68d6a6b --- /dev/null +++ b/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 = ( + '' + '' + '' + '' + '' + '' + '' + '' + '' + '' + ) + b64 = base64.b64encode(svg.encode("utf-8")).decode("ascii") + return f"data:image/svg+xml;base64,{b64}"