import logging from django.conf import settings from django.contrib.auth import get_user_model from django.utils import timezone from drf_spectacular.utils import extend_schema from rest_framework.permissions import AllowAny, IsAuthenticated from rest_framework.views import APIView from apps.audit_logs.models import AuthEventLog from apps.audit_logs.services import log_auth_event, log_activity from core.responses import error_response, success_response from . import mcp_client from .models import LoginAttempt from .serializers import LoginSerializer, MFAEnableSerializer, VerifyMFASerializer from .services import issue_tokens User = get_user_model() logger = logging.getLogger(__name__) def _login_error(message, code, status_code): """Error response for a failed login. `code` is a stable machine-readable identifier the frontend maps to the user-facing toast, so the wording can change without breaking the client. """ return error_response(message, errors={"code": code}, status_code=status_code) def _ensure_local_user(email, role="ADMIN"): user = User.objects.filter(email=email).first() if not user: user = User(email=email, username=email, role=role) user.set_unusable_password() user.save() return user class LoginView(APIView): authentication_classes = [] permission_classes = [AllowAny] @extend_schema(request=LoginSerializer, responses={200: None}) def post(self, request): # Validate manually so empty/malformed email and password can be reported # separately instead of as one lumped-together 400. raw_email = (request.data.get("email") or "").strip() raw_password = request.data.get("password") or "" if not raw_email and not raw_password: return _login_error("Please enter your email and password.", "missing_credentials", 400) if not raw_email: return _login_error("Please enter your email address.", "missing_email", 400) if not raw_password: return _login_error("Please enter your password.", "missing_password", 400) serializer = LoginSerializer(data={"email": raw_email, "password": raw_password}) if not serializer.is_valid(): if "email" in serializer.errors: return _login_error( "Please enter a valid email address.", "invalid_email_format", 400 ) return _login_error("Invalid email or password.", "invalid_credentials", 400) email = serializer.validated_data["email"] password = serializer.validated_data["password"] local_user = User.objects.filter(email=email).first() if local_user: if not local_user.is_active: log_auth_event(local_user, AuthEventLog.EventType.LOGIN_FAIL, request) return _login_error( "Your account has been disabled. Please contact the administrator.", "account_disabled", 403, ) attempt, _ = LoginAttempt.objects.get_or_create(user=local_user) if attempt.is_currently_locked(): log_auth_event(local_user, AuthEventLog.EventType.LOCKOUT, request) return _login_error( "Account locked due to too many failed attempts. Please contact the administrator.", "account_locked", 403, ) # 1) password check -> auth_mcp.verify_login # If auth_mcp is down we can still fall back to the local password hash; # only if BOTH paths are unavailable is this a genuine server error. mcp_down = False res = {} try: res = mcp_client.verify_login(email, password) except mcp_client.MCPServiceError as exc: mcp_down = True logger.error("auth_mcp unavailable during login for %s: %s", email, exc) if not isinstance(res, dict): res = {} verified = bool(res.get("verified")) mcp_exists = bool(res.get("exists", False)) # self-heal: if not in auth_mcp yet but valid locally, sync the hash then accept if not verified and local_user and local_user.check_password(password): if not mcp_down: try: mcp_client.sync_user(email, local_user.password, getattr(local_user, "role", "ADMIN")) except mcp_client.MCPServiceError as exc: logger.error("auth_mcp sync failed for %s: %s", email, exc) verified = True mcp_exists = True if not verified: if mcp_down and not local_user: # Can't tell whether the credentials were valid — don't imply they weren't. return _login_error( "Unable to log in at the moment. Please try again later.", "service_unavailable", 503, ) if local_user: attempt.register_failure() log_auth_event(local_user, AuthEventLog.EventType.LOGIN_FAIL, request) # `exists` is authoritative for "is this email registered": a local # user row or an auth_mcp record both count. email_registered = bool(local_user) or mcp_exists if not settings.LOGIN_REVEAL_ACCOUNT_EXISTENCE: return _login_error("Invalid email or password.", "invalid_credentials", 401) if not email_registered: return _login_error( "No account found with this email address.", "email_not_registered", 401 ) return _login_error( "Incorrect password. Please try again.", "incorrect_password", 401 ) user = _ensure_local_user(email, res.get("role") or "ADMIN") if not user.is_active: log_auth_event(user, AuthEventLog.EventType.LOGIN_FAIL, request) return _login_error( "Your account has been disabled. Please contact the administrator.", "account_disabled", 403, ) LoginAttempt.objects.get_or_create(user=user)[0].reset() # 2) MFA -> auth_mcp (central policy), keyed by email try: policy = mcp_client.get_mfa_policy().get("method", "none") if policy == "email": mcp_client.mfa_setup(email, email) # auth_mcp emails the code return success_response({"mfa_required": True, "method": "email"}, "MFA required. We emailed you a verification code.") if policy == "totp": status = mcp_client.mfa_status(email) if status.get("enabled") and status.get("enrolled"): return success_response({"mfa_required": True, "method": "totp"}, "MFA required. Please enter your authenticator code.") except mcp_client.MCPServiceError as exc: # Credentials were correct but the MFA step can't be reached — fail # closed rather than handing out tokens that skip MFA. logger.error("auth_mcp unavailable during MFA step for %s: %s", email, exc) return _login_error( "Unable to log in at the moment. Please try again later.", "service_unavailable", 503, ) if policy == "totp": log_auth_event(user, AuthEventLog.EventType.LOGIN_SUCCESS, request) log_activity(user, "USER_LOGIN", f"User {user.email} logged in (needs TOTP setup)", request) return success_response({**issue_tokens(user), "needs_totp_setup": True}, "Login successful. Please set up your authenticator.") log_auth_event(user, AuthEventLog.EventType.LOGIN_SUCCESS, request) log_activity(user, "USER_LOGIN", f"User {user.email} logged in", request) user.last_login = timezone.now(); user.save(update_fields=["last_login"]) return success_response(issue_tokens(user), "Login successful") class ResendOTPView(APIView): authentication_classes = [] permission_classes = [AllowAny] def post(self, request): if mcp_client.get_mfa_policy().get("method") != "email": return error_response("Resend is only for email OTP", status_code=400) email = request.data.get("email") if not email: return error_response("Email required", status_code=400) mcp_client.mfa_send_login_otp(email, email) return success_response(message="A new code has been emailed.") class VerifyMFAView(APIView): authentication_classes = [] permission_classes = [AllowAny] @extend_schema(request=VerifyMFASerializer, responses={200: None}) def post(self, request): serializer = VerifyMFASerializer(data=request.data) serializer.is_valid(raise_exception=True) email = serializer.validated_data["email"] code = serializer.validated_data["mfa_code"] user = User.objects.filter(email=email).first() if not user: return error_response("Invalid request", status_code=400) if not user.is_active: log_auth_event(user, AuthEventLog.EventType.LOGIN_FAIL, request) return error_response( "Your account is inactive. Please contact the administrator.", status_code=403, ) if not mcp_client.mfa_verify(email, code): log_auth_event(user, AuthEventLog.EventType.MFA_FAIL, request) policy = mcp_client.get_mfa_policy().get("method") msg = ("Invalid or expired email code. Please check your email." if policy == "email" else "Invalid authenticator code. Open your app and try again.") return error_response(msg, status_code=401) user.mfa_enabled = True user.mfa_method = mcp_client.get_mfa_policy().get("method", "none") user.last_login = timezone.now() user.save(update_fields=["mfa_enabled", "mfa_method", "last_login"]) log_auth_event(user, AuthEventLog.EventType.LOGIN_SUCCESS, request) log_activity(user, "USER_LOGIN", f"User {user.email} logged in via MFA", request) return success_response(issue_tokens(user), "Login successful") class MFAMethodsView(APIView): permission_classes = [IsAuthenticated] def get(self, request): method = mcp_client.get_mfa_policy().get("method", "email") return success_response({ "totp": method == "totp", "email": method == "email", "sms": method == "sms", "policy": method, "current": request.user.mfa_method, "enabled": request.user.mfa_enabled, }) class MFASetupView(APIView): permission_classes = [IsAuthenticated] def get(self, request): res = mcp_client.mfa_setup(request.user.email, request.user.email) if res.get("method") == "totp": return success_response({"secret": res["secret"], "qr_code": res["qr_code"]}, "Scan the QR code with your authenticator app") return success_response({"email": request.user.email, "sent": res.get("sent")}, "We emailed a verification code") class MFAEnableView(APIView): permission_classes = [IsAuthenticated] @extend_schema(request=MFAEnableSerializer, responses={200: None}) def post(self, request): serializer = MFAEnableSerializer(data=request.data) serializer.is_valid(raise_exception=True) if not mcp_client.mfa_verify(request.user.email, serializer.validated_data["code"]): return error_response("Invalid verification code", status_code=400) request.user.mfa_enabled = True request.user.mfa_method = mcp_client.get_mfa_policy().get("method", "none") request.user.save(update_fields=["mfa_enabled", "mfa_method"]) return success_response(message="MFA enabled successfully") class MFADisableView(APIView): permission_classes = [IsAuthenticated] def post(self, request): if getattr(request.user, "role", None) != "ADMIN": return error_response("Only an administrator can disable MFA", status_code=403) target_email = request.data.get("email") user = User.objects.filter(email=target_email).first() if target_email else request.user if not user: return error_response("User not found", status_code=404) user.mfa_enabled = False user.mfa_method = "none" user.save(update_fields=["mfa_enabled", "mfa_method"]) return success_response(message=f"MFA disabled for {user.email}")