done last commit before merging - fixed media URLSs S3
This commit is contained in:
@@ -6,7 +6,10 @@ from .serializers import *
|
||||
from .permissions import *
|
||||
from .models import CustomUser
|
||||
from .tokens import *
|
||||
from .tasks import send_password_reset_email_task, send_email_verification_task
|
||||
from .tasks import (
|
||||
send_password_reset_email_task, send_email_verification_task,
|
||||
send_email_change_verification_task, send_email_change_notification_task,
|
||||
)
|
||||
from django.conf import settings
|
||||
import logging
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -431,6 +434,208 @@ class PasswordResetConfirmView(APIView):
|
||||
user.save()
|
||||
return Response({"detail": "Heslo bylo úspěšně změněno."})
|
||||
return Response(serializer.errors, status=400)
|
||||
|
||||
|
||||
|
||||
@extend_schema(
|
||||
tags=["account"],
|
||||
summary="Resend email verification",
|
||||
description="Resends the verification email to the currently authenticated user. Limited to once per minute.",
|
||||
responses={
|
||||
200: OpenApiResponse(description="Verification email sent."),
|
||||
400: OpenApiResponse(description="Email is already verified."),
|
||||
429: OpenApiResponse(description="Rate limited — seconds_remaining returned."),
|
||||
},
|
||||
)
|
||||
class ResendEmailVerificationView(APIView):
|
||||
permission_classes = [IsAuthenticated]
|
||||
COOLDOWN = 60 # seconds
|
||||
|
||||
def post(self, request):
|
||||
from django.core.cache import cache
|
||||
import time
|
||||
|
||||
user = request.user
|
||||
if user.email_verified:
|
||||
return Response({"detail": "E-mail je již ověřen."}, status=status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
cache_key = f"resend_verification_{user.id}"
|
||||
sent_at = cache.get(cache_key)
|
||||
if sent_at is not None:
|
||||
remaining = int(self.COOLDOWN - (time.time() - sent_at))
|
||||
if remaining > 0:
|
||||
return Response(
|
||||
{"detail": "Příliš mnoho pokusů.", "seconds_remaining": remaining},
|
||||
status=status.HTTP_429_TOO_MANY_REQUESTS,
|
||||
)
|
||||
|
||||
cache.set(cache_key, time.time(), timeout=self.COOLDOWN)
|
||||
try:
|
||||
send_email_verification_task.delay(user.id)
|
||||
except Exception as e:
|
||||
logger.error(f"Celery not available, using fallback. Error: {e}")
|
||||
send_email_verification_task(user.id)
|
||||
return Response({"detail": "Ověřovací e-mail byl odeslán."})
|
||||
|
||||
|
||||
@extend_schema(
|
||||
tags=["account"],
|
||||
summary="Request email address change",
|
||||
description="Validates current password + Turnstile, stores pending_email, sends confirmation link to the new address and a notification to the old one.",
|
||||
responses={
|
||||
200: OpenApiResponse(description="Confirmation email sent to new address."),
|
||||
400: OpenApiResponse(description="Wrong password, duplicate email, or failed CAPTCHA."),
|
||||
},
|
||||
)
|
||||
class ChangeEmailView(APIView):
|
||||
permission_classes = [IsAuthenticated]
|
||||
|
||||
def post(self, request):
|
||||
turnstile_token = request.data.get("turnstile_token", "")
|
||||
remote_ip = request.META.get("HTTP_X_FORWARDED_FOR", request.META.get("REMOTE_ADDR", ""))
|
||||
if not verify_turnstile(turnstile_token, remote_ip):
|
||||
return Response({"detail": "Ověření CAPTCHA selhalo."}, status=status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
serializer = ChangeEmailSerializer(data=request.data)
|
||||
serializer.is_valid(raise_exception=True)
|
||||
|
||||
user = request.user
|
||||
if not user.check_password(serializer.validated_data["current_password"]):
|
||||
return Response({"current_password": "Nesprávné heslo."}, status=status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
new_email = serializer.validated_data["new_email"]
|
||||
old_email = user.email
|
||||
|
||||
user.pending_email = new_email
|
||||
uid = urlsafe_base64_encode(force_bytes(user.pk))
|
||||
token = user.generate_email_verification_token(save=False)
|
||||
user.save(update_fields=["pending_email", "email_verification_token", "email_verification_sent_at"])
|
||||
|
||||
verify_url = f"{settings.FRONTEND_URL}/account/confirm-email-change/{uid}/{token}/"
|
||||
|
||||
try:
|
||||
send_email_change_verification_task.delay(user.id, new_email, verify_url)
|
||||
send_email_change_notification_task.delay(user.id, old_email, new_email)
|
||||
except Exception as e:
|
||||
logger.error(f"Celery not available, using fallback. Error: {e}")
|
||||
send_email_change_verification_task(user.id, new_email, verify_url)
|
||||
send_email_change_notification_task(user.id, old_email, new_email)
|
||||
|
||||
return Response({"detail": "Ověřovací e-mail byl odeslán na novou adresu."})
|
||||
|
||||
|
||||
@extend_schema(
|
||||
tags=["account"],
|
||||
summary="Confirm email address change via link",
|
||||
description="Verifies uid + token from the confirmation email, swaps email to pending_email.",
|
||||
parameters=[
|
||||
OpenApiParameter(name="uidb64", type=str, location=OpenApiParameter.PATH),
|
||||
OpenApiParameter(name="token", type=str, location=OpenApiParameter.PATH),
|
||||
],
|
||||
responses={
|
||||
200: OpenApiResponse(description="Email changed successfully."),
|
||||
400: OpenApiResponse(description="Invalid or expired link."),
|
||||
},
|
||||
)
|
||||
class ConfirmEmailChangeView(APIView):
|
||||
def get(self, request, uidb64, token):
|
||||
ip = request.META.get("HTTP_X_FORWARDED_FOR", request.META.get("REMOTE_ADDR", "unknown"))
|
||||
try:
|
||||
uid = force_str(urlsafe_base64_decode(uidb64))
|
||||
user = User.objects.get(pk=uid)
|
||||
except (User.DoesNotExist, ValueError, TypeError):
|
||||
return Response({"error": "Neplatný odkaz."}, status=status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
if not user.pending_email:
|
||||
return Response({"error": "Žádná čekající změna e-mailu."}, status=status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
if not user.verify_email_token(token):
|
||||
return Response({"error": "Token je neplatný nebo expirovaný."}, status=status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
old_email = user.email
|
||||
user.email = user.pending_email
|
||||
user.pending_email = None
|
||||
user.email_verified = True
|
||||
user.save(update_fields=["email", "pending_email", "email_verified"])
|
||||
logger.info("Email changed for user %s: %s → %s (IP: %s)", user.pk, old_email, user.email, ip)
|
||||
|
||||
return Response({"detail": "E-mail byl úspěšně změněn."})
|
||||
|
||||
|
||||
@extend_schema(
|
||||
tags=["account"],
|
||||
summary="Change username",
|
||||
description="Updates the username. Requires Turnstile. Limited to once per 30 days.",
|
||||
responses={
|
||||
200: OpenApiResponse(description="Username changed."),
|
||||
400: OpenApiResponse(description="Duplicate name, invalid characters, or failed CAPTCHA."),
|
||||
429: OpenApiResponse(description="Rate limited — days_remaining returned."),
|
||||
},
|
||||
)
|
||||
class ChangeUsernameView(APIView):
|
||||
permission_classes = [IsAuthenticated]
|
||||
COOLDOWN_DAYS = 30
|
||||
|
||||
def post(self, request):
|
||||
from django.core.cache import cache
|
||||
import time
|
||||
|
||||
turnstile_token = request.data.get("turnstile_token", "")
|
||||
remote_ip = request.META.get("HTTP_X_FORWARDED_FOR", request.META.get("REMOTE_ADDR", ""))
|
||||
if not verify_turnstile(turnstile_token, remote_ip):
|
||||
return Response({"detail": "Ověření CAPTCHA selhalo."}, status=status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
serializer = ChangeUsernameSerializer(data=request.data)
|
||||
serializer.is_valid(raise_exception=True)
|
||||
|
||||
user = request.user
|
||||
cache_key = f"username_change_{user.id}"
|
||||
changed_at = cache.get(cache_key)
|
||||
if changed_at is not None:
|
||||
remaining_days = int(self.COOLDOWN_DAYS - (time.time() - changed_at) / 86400)
|
||||
if remaining_days > 0:
|
||||
return Response(
|
||||
{"detail": "Příliš brzy.", "days_remaining": remaining_days},
|
||||
status=status.HTTP_429_TOO_MANY_REQUESTS,
|
||||
)
|
||||
|
||||
user.username = serializer.validated_data["new_username"]
|
||||
user.save(update_fields=["username"])
|
||||
cache.set(cache_key, time.time(), timeout=self.COOLDOWN_DAYS * 86400)
|
||||
|
||||
return Response({"detail": "Uživatelské jméno bylo změněno.", "username": user.username})
|
||||
|
||||
|
||||
@extend_schema(
|
||||
tags=["account"],
|
||||
summary="Delete own account",
|
||||
description="Soft-deletes the authenticated user's account after password confirmation. Clears auth cookies.",
|
||||
responses={
|
||||
204: OpenApiResponse(description="Account deleted."),
|
||||
400: OpenApiResponse(description="Wrong password."),
|
||||
},
|
||||
)
|
||||
class DeleteAccountView(APIView):
|
||||
permission_classes = [IsAuthenticated]
|
||||
|
||||
def post(self, request):
|
||||
user = request.user
|
||||
|
||||
turnstile_token = request.data.get("turnstile_token", "")
|
||||
remote_ip = request.META.get("HTTP_X_FORWARDED_FOR", request.META.get("REMOTE_ADDR", ""))
|
||||
if not verify_turnstile(turnstile_token, remote_ip):
|
||||
return Response({"detail": "Ověření CAPTCHA selhalo."}, status=status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
password = request.data.get("current_password", "")
|
||||
if not password or not user.check_password(password):
|
||||
return Response({"current_password": "Nesprávné heslo."}, status=status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
logger.info("Account deletion requested for user %s (id=%s)", user.username, user.pk)
|
||||
django_logout(request)
|
||||
user.delete()
|
||||
|
||||
response = Response(status=status.HTTP_204_NO_CONTENT)
|
||||
response.delete_cookie("access_token", path="/")
|
||||
response.delete_cookie("refresh_token", path="/")
|
||||
return response
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user