import secrets import logging from django.conf import settings from django.contrib.auth import get_user_model, authenticate from django.contrib.auth.password_validation import validate_password from django.contrib.auth.tokens import default_token_generator from django.core.exceptions import ValidationError from django.core.mail import send_mail from django.utils import timezone from django.utils.encoding import force_bytes, force_str from django.utils.http import urlsafe_base64_encode, urlsafe_base64_decode from rest_framework import status, generics, permissions from rest_framework.response import Response from rest_framework.views import APIView from rest_framework.throttling import AnonRateThrottle from rest_framework_simplejwt.tokens import RefreshToken from .serializers import RegistrationSerializer, LoginSerializer, VerifyOTPSerializer, UserSerializer import requests User = get_user_model() logger = logging.getLogger(__name__) def verify_turnstile(token): secret_key = getattr(settings, 'TURNSTILE_SECRET_KEY', None) if not secret_key: return True # Ja atslēga nav konfigurēta, izlaižam pārbaudi if not token: return False try: response = requests.post('https://challenges.cloudflare.com/turnstile/v0/siteverify', data={ 'secret': secret_key, 'response': token }, timeout=5) return response.json().get('success', False) except Exception as e: logger.error(f"Turnstile kļūda: {e}") return False def set_jwt_cookies(response, access_token, refresh_token=None): cookie_name = getattr(settings, 'SIMPLE_JWT', {}).get('AUTH_COOKIE', 'access_token') response.set_cookie( cookie_name, access_token, httponly=True, secure=True, # Jābūt True, ja SameSite=None samesite='None', # Atļauj cross-domain sīkdatnes max_age=30 * 60 # 30 minūtes ) if refresh_token: response.set_cookie( 'refresh_token', refresh_token, httponly=True, secure=True, samesite='None', max_age=10 * 60 * 60 # 10 stundas ) return response # === CUSTOM THROTTLES (DROŠĪBA: Rate Limiting) === class LoginThrottle(AnonRateThrottle): rate = '10/hour' class OTPThrottle(AnonRateThrottle): rate = '5/hour' class PasswordResetThrottle(AnonRateThrottle): rate = '3/hour' # --- 1. Reģistrācija --- class RegisterView(APIView): def post(self, request): turnstile_token = request.data.get('turnstile_token') if not verify_turnstile(turnstile_token): return Response({"error": "CAPTCHA validācija neizdevās. Lūdzu, mēģiniet vēlreiz."}, status=status.HTTP_400_BAD_REQUEST) serializer = RegistrationSerializer(data=request.data) if serializer.is_valid(): user = serializer.save() user.refresh_from_db() token = default_token_generator.make_token(user) uid = urlsafe_base64_encode(force_bytes(user.pk)) activation_link = f"{settings.FRONTEND_URL}/confirm-email/{uid}/{token}/" subject = "Apstiprini savu MiM kontu" message = f"Sveiki, {getattr(user, 'full_name', 'Lietotāj')}!\n\nLūdzu, apstipriniet reģistrāciju:\n{activation_link}" try: send_mail(subject, message, settings.EMAIL_HOST_USER, [user.email]) return Response({"msg": "REĢISTRĀCIJA_VEIKSMĪGA. Pārbaudi e-pastu aktivizācijai."}, status=status.HTTP_201_CREATED) except Exception as e: # MED-03: Neatklāt iekšējo kļūdu klientam logger.error(f"Reģistrācija: e-pasta kļūda lietotājam {user.email}: {e}") user.delete() return Response({"error": "Neizdevās nosūtīt aktivizācijas e-pastu. Mēģiniet vēlreiz."}, status=status.HTTP_500_INTERNAL_SERVER_ERROR) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) # --- 2. E-pasta apstiprināšana --- class VerifyEmailView(APIView): def post(self, request): uid = request.data.get('uid') token = request.data.get('token') if not uid or not token: return Response({'error': 'Trūkst datu (uid vai token)'}, status=status.HTTP_400_BAD_REQUEST) try: uid_decoded = force_str(urlsafe_base64_decode(uid)) user = User.objects.get(pk=uid_decoded) except (TypeError, ValueError, OverflowError, User.DoesNotExist): return Response({'error': 'Nederīga saite'}, status=status.HTTP_400_BAD_REQUEST) if user.is_active: return Response({'msg': 'Konts jau ir aktivizēts. Vari ielogoties.'}, status=status.HTTP_200_OK) if default_token_generator.check_token(user, token): user.is_active = True user.save() return Response({'msg': 'VEIKSMĪGI! Konts aktivizēts. Tagad vari ielogoties.'}, status=status.HTTP_200_OK) else: return Response({'error': 'Saite ir novecojusi vai nederīga'}, status=status.HTTP_400_BAD_REQUEST) # --- 3. Login (ar 2FA) — DROŠĪBA: Rate Limiting + Crypto OTP --- class LoginView(APIView): permission_classes = (permissions.AllowAny,) throttle_classes = [LoginThrottle] # DROŠĪBA: Max 10 mēģinājumi/h def post(self, request): serializer = LoginSerializer(data=request.data) if serializer.is_valid(): email = serializer.validated_data['email'] password = serializer.validated_data['password'] user = authenticate(username=email, password=password) if user: if not user.is_active: return Response({"error": "Konts nav aktivizēts. Pārbaudi e-pastu!"}, status=status.HTTP_403_FORBIDDEN) # CRIT-05: Kriptogrāfiski drošs OTP otp = str(secrets.randbelow(900000) + 100000) # 100000-999999 user.otp_code = otp user.otp_created_at = timezone.now() # Laika zīmogs user.otp_attempts = 0 # Atiestatām mēģinājumu skaitītāju user.save() try: send_mail( "Tavs MiM Drošības Kods", f"Tavs verifikācijas kods ir: {otp}\n\nKods ir derīgs 5 minūtes.", settings.EMAIL_HOST_USER, [user.email] ) return Response({"requires_2fa": True, "msg": "Kods nosūtīts uz e-pastu"}, status=status.HTTP_200_OK) except Exception as e: logger.error(f"Login 2FA: e-pasta kļūda {user.email}: {e}") return Response({"error": "Neizdevās nosūtīt 2FA kodu. Mēģiniet vēlreiz."}, status=status.HTTP_500_INTERNAL_SERVER_ERROR) return Response({"error": "Nepareizs e-pasts vai parole"}, status=status.HTTP_401_UNAUTHORIZED) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) # --- 4. 2FA Pārbaude — DROŠĪBA: Laika limits + Mēģinājumu limits --- class Verify2FAView(APIView): permission_classes = (permissions.AllowAny,) throttle_classes = [OTPThrottle] # DROŠĪBA: Max 5 mēģinājumi/h def post(self, request): serializer = VerifyOTPSerializer(data=request.data) if serializer.is_valid(): email = serializer.validated_data['email'] otp = serializer.validated_data['otp'] try: user = User.objects.get(email=email) # CRIT-05: Pārbaudīt mēģinājumu limitu max_attempts = getattr(settings, 'OTP_MAX_ATTEMPTS', 5) if hasattr(user, 'otp_attempts') and user.otp_attempts and user.otp_attempts >= max_attempts: user.otp_code = None # Anulējam kodu user.save() return Response({"error": "Pārsniegts mēģinājumu limits. Pieslēdzieties vēlreiz."}, status=status.HTTP_429_TOO_MANY_REQUESTS) # CRIT-05: Pārbaudīt derīguma termiņu (5 min) otp_expiry = getattr(settings, 'OTP_EXPIRY_SECONDS', 300) if hasattr(user, 'otp_created_at') and user.otp_created_at: elapsed = (timezone.now() - user.otp_created_at).total_seconds() if elapsed > otp_expiry: user.otp_code = None user.save() return Response({"error": "Kods ir beidzies. Pieslēdzieties vēlreiz, lai saņemtu jaunu kodu."}, status=status.HTTP_400_BAD_REQUEST) if user.otp_code == otp: user.otp_code = None if hasattr(user, 'otp_attempts'): user.otp_attempts = 0 user.save() refresh = RefreshToken.for_user(user) response_data = { 'user': { 'id': user.id, 'full_name': getattr(user, 'full_name', ''), 'email': user.email, 'role': getattr(user, 'role', 'user'), 'is_staff': user.is_staff, 'telegram_link_code': str(user.telegram_link_code) if user.telegram_link_code else None, } } response = Response(response_data, status=status.HTTP_200_OK) return set_jwt_cookies(response, str(refresh.access_token), str(refresh)) else: # Palielināt mēģinājumu skaitītāju if hasattr(user, 'otp_attempts'): user.otp_attempts = (user.otp_attempts or 0) + 1 user.save() return Response({"error": "Nepareizs 2FA kods"}, status=status.HTTP_400_BAD_REQUEST) except User.DoesNotExist: return Response({"error": "Lietotājs nav atrasts"}, status=status.HTTP_404_NOT_FOUND) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) # --- 5. GOOGLE LOGIN (CRIT-06: Izlabots — nav 2FA bypass, TOS pārbaude) --- class GoogleLoginView(APIView): permission_classes = [permissions.AllowAny] throttle_classes = [LoginThrottle] def post(self, request): access_token = request.data.get('access_token') if not access_token: return Response({"error": "Netika saņemts Google žetons."}, status=status.HTTP_400_BAD_REQUEST) # 1. Validējam Google žetonu try: google_response = requests.get( 'https://www.googleapis.com/oauth2/v3/userinfo', params={'access_token': access_token}, timeout=10 ) except requests.RequestException: return Response({"error": "Google serviss nav pieejams."}, status=status.HTTP_503_SERVICE_UNAVAILABLE) if not google_response.ok: return Response({"error": "Nederīgs vai beidzies Google žetons."}, status=status.HTTP_400_BAD_REQUEST) google_data = google_response.json() email = google_data.get('email') full_name = google_data.get('name', '') if not email: return Response({"error": "Google neatgrieza e-pasta adresi."}, status=status.HTTP_400_BAD_REQUEST) # 2. Atrodam vai izveidojam lietotāju user, created = User.objects.get_or_create(email=email) if created: user.full_name = full_name user.set_unusable_password() user.is_active = True user.save() # CRIT-06: NEaktivizēt manuāli deaktivētu kontu elif not user.is_active: return Response({"error": "Konts ir deaktivēts. Sazinieties ar atbalstu."}, status=status.HTTP_403_FORBIDDEN) # 3. Ģenerējam JWT žetonus (Google lietotāji izlaiž 2FA, jo Google jau veic savu verifikāciju) refresh = RefreshToken.for_user(user) response_data = { 'user': { 'id': user.id, 'email': user.email, 'full_name': getattr(user, 'full_name', ''), 'is_staff': user.is_staff, 'telegram_link_code': str(user.telegram_link_code) if user.telegram_link_code else None, # CRIT-06: Informēt frontend, vai TOS ir apstiprināti 'terms_accepted': user.terms_accepted, } } response = Response(response_data, status=status.HTTP_200_OK) return set_jwt_cookies(response, str(refresh.access_token), str(refresh)) # --- 6. Profils --- class UserDetailView(generics.RetrieveAPIView): serializer_class = UserSerializer permission_classes = [permissions.IsAuthenticated] def get_object(self): return self.request.user # --- LOGOUT --- class LogoutView(APIView): permission_classes = [permissions.AllowAny] def post(self, request): response = Response({"msg": "Izrakstīts veiksmīgi"}, status=status.HTTP_200_OK) cookie_name = getattr(settings, 'SIMPLE_JWT', {}).get('AUTH_COOKIE', 'access_token') response.delete_cookie(cookie_name) response.delete_cookie('refresh_token') return response # --- TOKEN REFRESH --- class CookieTokenRefreshView(APIView): permission_classes = [permissions.AllowAny] throttle_classes = [OTPThrottle] def post(self, request): refresh_token = request.COOKIES.get('refresh_token') if not refresh_token: return Response({"error": "Trūkst atjaunošanas žetona"}, status=status.HTTP_401_UNAUTHORIZED) try: refresh = RefreshToken(refresh_token) access_token = str(refresh.access_token) response = Response({"msg": "Žetons atjaunots"}, status=status.HTTP_200_OK) return set_jwt_cookies(response, access_token) except Exception as e: return Response({"error": "Nederīgs atjaunošanas žetons"}, status=status.HTTP_401_UNAUTHORIZED) # --- 7. Paroles Atiestatīšana (MED-04: Rate Limiting + MED-07: Pilna validācija) --- class PasswordResetRequestView(APIView): permission_classes = (permissions.AllowAny,) throttle_classes = [PasswordResetThrottle] # DROŠĪBA: Max 3/h def post(self, request): email = request.data.get("email", "").strip().lower() if not email: return Response({"error": "E-pasta adrese ir obligāta."}, status=status.HTTP_400_BAD_REQUEST) try: user = User.objects.get(email=email) token = default_token_generator.make_token(user) uid = urlsafe_base64_encode(force_bytes(user.pk)) reset_link = f"{settings.FRONTEND_URL}/reset-password/{uid}/{token}/" subject = "MIM — Paroles Atiestatīšana" message = ( f"Sveiki!\n\nPieprasīta paroles atiestatīšana MIM kontam.\n" f"Saite (derīga 24h):\n{reset_link}\n\n" f"Ja Tu to nepieprasīji, ignorē šo ziņu." ) send_mail(subject, message, settings.EMAIL_HOST_USER, [user.email]) except User.DoesNotExist: pass # Drošība: neatklāj, vai e-pasts eksistē except Exception as e: # MED-03: Slēpjam iekšējo kļūdu logger.error(f"Password reset e-pasta kļūda: {e}") return Response({"msg": "Instrukcijas nosūtītas, ja e-pasts eksistē."}, status=status.HTTP_200_OK) class PasswordResetConfirmView(APIView): permission_classes = (permissions.AllowAny,) def post(self, request): uid = request.data.get("uid", "") token = request.data.get("token", "") new_pass = request.data.get("new_password", "") if not uid or not token or not new_pass: return Response({"error": "Trūkst datu."}, status=status.HTTP_400_BAD_REQUEST) try: uid_decoded = force_str(urlsafe_base64_decode(uid)) user = User.objects.get(pk=uid_decoded) except (TypeError, ValueError, OverflowError, User.DoesNotExist): return Response({"error": "Nederīgs lietotājs."}, status=status.HTTP_400_BAD_REQUEST) if not default_token_generator.check_token(user, token): return Response({"error": "Tokenis nederīgs vai novecojis."}, status=status.HTTP_400_BAD_REQUEST) # MED-07: Pilna Django paroles validācija (ne tikai garums!) try: validate_password(new_pass, user) except ValidationError as e: return Response({"error": list(e.messages)}, status=status.HTTP_400_BAD_REQUEST) user.set_password(new_pass) user.save() return Response({"msg": "Parole nomainīta!"}, status=status.HTTP_200_OK)