"""Session-cookie auth endpoints. Endpoints mounted under ``/api/v1/auth/``: * ``POST /login/`` — authenticate by mobile + password, open a session. * ``POST /logout/`` — drop the session. * ``POST /refresh/`` — touch the session. * ``GET /me/`` — return the ``SessionContext`` for the SPA. * ``GET /csrf/`` — seed the ``csrftoken`` cookie. * ``POST /otp/request/`` — send a one-time SMS code to a mobile number. * ``POST /otp/verify/`` — verify the OTP code and open a session. All endpoints are exempt from the PermissionGateMiddleware because they bootstrap the auth state. """ from __future__ import annotations import random import string from datetime import timedelta from typing import Any from django.contrib.auth import authenticate, login, logout from django.contrib.auth import get_user_model from django.db.models import Q from django.utils import timezone from django.views.decorators.csrf import ensure_csrf_cookie from rest_framework import status from rest_framework.decorators import api_view, permission_classes from rest_framework.permissions import AllowAny, IsAuthenticated from rest_framework.request import Request from rest_framework.response import Response from simorgh.apps.iam.engine import collect_permissions_for_roles from simorgh.apps.memberships.services import active_memberships_for, assignable_node_ids from .serializers import LoginSerializer, OtpRequestSerializer, OtpVerifySerializer _OTP_TTL_MINUTES = 5 _OTP_MAX_ATTEMPTS = 5 _OTP_DAILY_LIMIT = 10 def _generate_otp_code(length: int = 6) -> str: return "".join(random.choices(string.digits, k=length)) def _session_context(request: Request) -> dict[str, Any]: """Build the ``SessionContext`` shape consumed by the SPA.""" user = request.user tenant = getattr(request, "tenant", None) permissions: list[str] = [] org_node_ids: list[int] = [] enabled_features: list[str] = [] avatar_url: str | None = None if user.is_authenticated and tenant is not None: memberships = list(active_memberships_for(user.pk, tenant.pk)) roles = [m.role for m in memberships] permissions = sorted(collect_permissions_for_roles(roles)) org_node_ids = sorted(assignable_node_ids(memberships)) # Resolve subscription-enabled features for frontend gating from simorgh.apps.subscription.selectors import get_all_enabled_features enabled_features = sorted(get_all_enabled_features(tenant.pk)) # Resolve avatar URL if user.is_authenticated: profile = getattr(user, "profile", None) if profile is not None and profile.avatar_image: avatar_url = profile.avatar_image.url return { "actor": { "id": user.pk, "email": user.email, "mobile": getattr(user, "mobile", None), "fullName": (user.get_full_name() or user.username or user.email or getattr(user, "mobile", None)), "avatarUrl": avatar_url, } if user.is_authenticated else None, "tenant": { "id": tenant.pk, "slug": tenant.slug, "name": tenant.name, } if tenant is not None else None, "permissions": permissions, "orgNodeIds": org_node_ids, "enabledFeatures": enabled_features, "isSuperuser": bool(user.is_authenticated and user.is_superuser), } @api_view(["POST"]) @permission_classes([AllowAny]) def login_view(request: Request) -> Response: serializer = LoginSerializer(data=request.data) serializer.is_valid(raise_exception=True) identifier = serializer.validated_data["identifier"] password = serializer.validated_data["password"] UserModel = get_user_model() candidate = ( UserModel.objects.filter( Q(email__iexact=identifier) | Q(username=identifier) | Q(mobile=identifier) ) .first() ) # USERNAME_FIELD is now "mobile"; pass it to ModelBackend.authenticate(). auth_identifier = candidate.mobile if candidate is not None else identifier user = authenticate(request, username=auth_identifier, password=password) if user is None or not user.is_active: return Response( {"error": {"code": "invalid_credentials", "message": "Invalid mobile or password."}}, status=status.HTTP_401_UNAUTHORIZED, ) login(request, user) from simorgh.apps.tenants.resolver import resolve_tenant request.tenant = resolve_tenant(request) return Response(_session_context(request), status=status.HTTP_200_OK) @api_view(["POST"]) @permission_classes([AllowAny]) def logout_view(request: Request) -> Response: logout(request) return Response(status=status.HTTP_204_NO_CONTENT) @api_view(["POST"]) @permission_classes([IsAuthenticated]) def refresh_view(request: Request) -> Response: request.session.modified = True return Response(status=status.HTTP_204_NO_CONTENT) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def me_view(request: Request) -> Response: return Response(_session_context(request)) @api_view(["GET"]) @permission_classes([AllowAny]) @ensure_csrf_cookie def csrf_view(_request: Request) -> Response: return Response(status=status.HTTP_204_NO_CONTENT) # --------------------------------------------------------------------------- # Mobile OTP endpoints # --------------------------------------------------------------------------- @api_view(["POST"]) @permission_classes([AllowAny]) def otp_request_view(request: Request) -> Response: """Generate and send a 6-digit OTP to the given mobile number. Rate-limited: at most ``_OTP_DAILY_LIMIT`` requests per mobile per day. Returns ``{"token": ""}`` which the client passes to the verify endpoint. """ from simorgh.apps.accounts.models import MobileOtpToken serializer = OtpRequestSerializer(data=request.data) serializer.is_valid(raise_exception=True) mobile: str = serializer.validated_data["mobile"] # Daily rate limit (simple DB count — no Redis dependency for OTP) today_start = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0) daily_count = MobileOtpToken.objects.filter( mobile=mobile, created_at__gte=today_start ).count() if daily_count >= _OTP_DAILY_LIMIT: return Response( {"error": {"code": "rate_limit", "message": "تعداد درخواست‌های مجاز امروز پر شده است."}}, status=status.HTTP_429_TOO_MANY_REQUESTS, ) code = _generate_otp_code() expires_at = timezone.now() + timedelta(minutes=_OTP_TTL_MINUTES) tenant = getattr(request, "tenant", None) token_obj = MobileOtpToken.objects.create( mobile=mobile, code=code, expires_at=expires_at, tenant=tenant, ) # Send via SMS provider system send_status = "sent" send_error = "" send_provider = "" try: from simorgh.apps.notifications.sms import get_provider_for_mobile provider = get_provider_for_mobile(mobile) if provider is not None: send_provider = getattr(provider, "name", type(provider).__name__) provider.send_otp(mobile=mobile, code=code, tenant_id=getattr(tenant, "pk", None)) except Exception as exc: # noqa: BLE001 send_status = "failed" send_error = str(exc) import logging logging.getLogger(__name__).exception("Failed to send OTP to %s", mobile) # Audit log — record every send attempt. from simorgh.apps.accounts.models import OtpSendLog ip = request.META.get("HTTP_X_FORWARDED_FOR", request.META.get("REMOTE_ADDR", "")) ip = ip.split(",")[0].strip() or None OtpSendLog.objects.create( mobile=mobile, otp_token=token_obj, tenant=tenant, provider=send_provider, status=send_status, error_message=send_error, ip_address=ip, ) return Response( {"token": str(token_obj.public_id), "expires_in": _OTP_TTL_MINUTES * 60}, status=status.HTTP_200_OK, ) @api_view(["POST"]) @permission_classes([AllowAny]) def otp_verify_view(request: Request) -> Response: """Verify an OTP code and open a session on success.""" from simorgh.apps.accounts.models import MobileOtpToken serializer = OtpVerifySerializer(data=request.data) serializer.is_valid(raise_exception=True) token_id = serializer.validated_data["token"] code: str = serializer.validated_data["code"] mobile: str = serializer.validated_data["mobile"] try: token_obj = MobileOtpToken.objects.get( public_id=token_id, mobile=mobile, is_used=False, expires_at__gt=timezone.now(), ) except MobileOtpToken.DoesNotExist: return Response( {"error": {"code": "invalid_token", "message": "کد نامعتبر یا منقضی شده است."}}, status=status.HTTP_400_BAD_REQUEST, ) token_obj.attempts += 1 if token_obj.attempts > _OTP_MAX_ATTEMPTS: token_obj.save() return Response( {"error": {"code": "too_many_attempts", "message": "تعداد تلاش‌های مجاز تمام شد."}}, status=status.HTTP_429_TOO_MANY_REQUESTS, ) if token_obj.code != code: token_obj.save() return Response( {"error": {"code": "wrong_code", "message": "کد وارد شده اشتباه است."}}, status=status.HTTP_400_BAD_REQUEST, ) # Mark token as used token_obj.is_used = True token_obj.save() # Find the user by mobile UserModel = get_user_model() try: user = UserModel.objects.get(mobile=mobile, is_active=True) except UserModel.DoesNotExist: return Response( {"error": {"code": "user_not_found", "message": "کاربری با این شماره موبایل یافت نشد."}}, status=status.HTTP_401_UNAUTHORIZED, ) login(request, user, backend="django.contrib.auth.backends.ModelBackend") from simorgh.apps.tenants.resolver import resolve_tenant request.tenant = resolve_tenant(request) return Response(_session_context(request), status=status.HTTP_200_OK) # --------------------------------------------------------------------------- # User profile endpoint # --------------------------------------------------------------------------- @api_view(["GET", "PATCH"]) @permission_classes([IsAuthenticated]) def profile_view(request: Request) -> Response: """GET or PATCH the current user's profile preferences. PATCH accepts: ``display_name``, ``language_id``, ``timezone_id``. """ user = request.user if request.method == "PATCH": data = request.data if "display_name" in data: user.username = str(data["display_name"])[:150] if "language_id" in data: from simorgh.apps.localization.models import Language lang_id = data["language_id"] if lang_id is None: user.language = None else: try: user.language = Language.objects.get(pk=int(lang_id), is_enabled=True) except (Language.DoesNotExist, ValueError, TypeError): return Response( {"error": {"code": "invalid_language", "message": "Invalid language."}}, status=status.HTTP_400_BAD_REQUEST, ) if "timezone_id" in data: from simorgh.apps.localization.models import Timezone tz_id = data["timezone_id"] if tz_id is None: user.timezone = None else: try: user.timezone = Timezone.objects.get(pk=int(tz_id), is_enabled=True) except (Timezone.DoesNotExist, ValueError, TypeError): return Response( {"error": {"code": "invalid_timezone", "message": "Invalid timezone."}}, status=status.HTTP_400_BAD_REQUEST, ) if "mobile_country_id" in data: from simorgh.apps.localization.models import Country mc_id = data["mobile_country_id"] if mc_id is None: user.mobile_country = None else: try: user.mobile_country = Country.objects.get(pk=int(mc_id), is_enabled=True) except (Country.DoesNotExist, ValueError, TypeError): return Response( {"error": {"code": "invalid_country", "message": "Invalid country."}}, status=status.HTTP_400_BAD_REQUEST, ) update_fields = ["username", "language", "timezone", "mobile_country"] user.save(update_fields=update_fields) language = user.language tz = user.timezone mc = getattr(user, "mobile_country", None) profile = getattr(user, "profile", None) avatar_url: str | None = None if profile is not None and profile.avatar_image: avatar_url = request.build_absolute_uri(profile.avatar_image.url) return Response( { "display_name": user.username, "email": user.email, "mobile": getattr(user, "mobile", None), "avatar_url": avatar_url, "mobile_country": { "id": mc.pk, "iso2": mc.iso2, "dial_code": mc.dial_code, "flag_emoji": mc.flag_emoji, "name_english": mc.name_english, } if mc is not None else None, "language": { "id": language.pk, "code": language.code, "name_english": language.name_english, "name_native": language.name_native, "direction": language.direction, } if language is not None else None, "timezone": { "id": tz.pk, "name": tz.name, "display_name": tz.display_name, "utc_offset": tz.utc_offset, } if tz is not None else None, } ) # --------------------------------------------------------------------------- # Avatar upload endpoint # --------------------------------------------------------------------------- AVATAR_MAX_BYTES = 5 * 1024 * 1024 # 5 MiB AVATAR_ALLOWED_CONTENT_TYPES = {"image/jpeg", "image/png", "image/gif", "image/webp"} @api_view(["POST"]) @permission_classes([IsAuthenticated]) def avatar_upload_view(request: Request) -> Response: """POST multipart/form-data with field ``avatar`` to update the user's avatar.""" avatar_file = request.FILES.get("avatar") if not avatar_file: return Response( {"error": {"code": "missing_file", "message": "No file uploaded."}}, status=status.HTTP_400_BAD_REQUEST, ) if avatar_file.size > AVATAR_MAX_BYTES: return Response( {"error": {"code": "file_too_large", "message": "File exceeds 5 MiB limit."}}, status=status.HTTP_400_BAD_REQUEST, ) content_type = getattr(avatar_file, "content_type", "") or "" if content_type not in AVATAR_ALLOWED_CONTENT_TYPES: return Response( {"error": {"code": "invalid_type", "message": "Only JPEG, PNG, GIF, and WebP are allowed."}}, status=status.HTTP_400_BAD_REQUEST, ) profile, _ = request.user.profile.__class__.objects.get_or_create(user=request.user) if profile.avatar_image: try: profile.avatar_image.delete(save=False) except Exception: # noqa: BLE001 pass profile.avatar_image = avatar_file profile.save(update_fields=["avatar_image"]) return Response({"avatar_url": request.build_absolute_uri(profile.avatar_image.url)}) # --------------------------------------------------------------------------- # Change password endpoint # --------------------------------------------------------------------------- @api_view(["POST"]) @permission_classes([IsAuthenticated]) def change_password_view(request: Request) -> Response: """Change the authenticated user's password. Accepts: ``current_password``, ``new_password``. """ current = request.data.get("current_password", "") new_password = request.data.get("new_password", "") if not current or not new_password: return Response( {"error": {"code": "missing_fields", "message": "Both current and new password are required."}}, status=status.HTTP_400_BAD_REQUEST, ) if len(new_password) < 6: return Response( {"error": {"code": "password_too_short", "message": "New password must be at least 6 characters."}}, status=status.HTTP_400_BAD_REQUEST, ) user = request.user if not user.check_password(current): return Response( {"error": {"code": "wrong_password", "message": "Current password is incorrect."}}, status=status.HTTP_400_BAD_REQUEST, ) user.set_password(new_password) user.save(update_fields=["password"]) # Re-login to update the session hash after password change from django.contrib.auth import update_session_auth_hash update_session_auth_hash(request, user) return Response({"message": "Password changed successfully."}) # --------------------------------------------------------------------------- # Tenant selection endpoints (multi-tenant browser flow) # --------------------------------------------------------------------------- @api_view(["GET"]) @permission_classes([IsAuthenticated]) def tenants_view(request: Request) -> Response: """List all tenants the authenticated user has an active membership in. Used by the tenant-selection page after login when the session has no active tenant yet. Returns a flat list suitable for display in the UI. """ from simorgh.apps.memberships.models import Membership from simorgh.apps.tenants.models import Tenant, TenantStatus # A user can belong to multiple memberships in the same tenant (different roles), # so deduplicate at the tenant level rather than returning one entry per membership. tenant_ids = ( Membership.objects.filter(users=request.user) .values_list("tenant_id", flat=True) .distinct() ) tenants_qs = ( Tenant.objects.filter(pk__in=tenant_ids, status=TenantStatus.ACTIVE) .order_by("name") ) return Response( [{"id": t.pk, "slug": t.slug, "name": t.name} for t in tenants_qs] ) @api_view(["POST"]) @permission_classes([IsAuthenticated]) def switch_tenant_view(request: Request) -> Response: """Persist a tenant selection in the server-side session. After this call the next ``GET /auth/me/`` (or any request without an explicit ``X-Tenant`` header) will resolve the tenant from the session, so the SPA can start sending tenant-scoped requests immediately. Returns the updated ``SessionContext`` so the client can ``applySession`` without a separate round-trip. """ from simorgh.apps.memberships.models import Membership from simorgh.apps.tenants.models import Tenant, TenantStatus slug = (request.data or {}).get("tenant") if not slug: return Response( {"error": {"code": "missing_tenant", "message": "tenant slug is required"}}, status=status.HTTP_400_BAD_REQUEST, ) tenant = Tenant.objects.filter(slug=slug, status=TenantStatus.ACTIVE).first() if tenant is None: return Response( {"error": {"code": "not_found", "message": "Tenant not found or inactive"}}, status=status.HTTP_404_NOT_FOUND, ) if not request.user.is_superuser: has_access = Membership.objects.filter(users=request.user, tenant=tenant).exists() if not has_access: return Response( {"error": {"code": "forbidden", "message": "You are not a member of this tenant"}}, status=status.HTTP_403_FORBIDDEN, ) request.session["active_tenant_slug"] = slug request.session.modified = True request.tenant = tenant return Response(_session_context(request), status=status.HTTP_200_OK)