Internal
Public Access
Keith SmithandClaude Sonnet 4.5
a03b9fe79c
Added rate limiting to previously unprotected user-related endpoints. Protected Endpoints: - UserProfileAPIView: Profile retrieval and updates (1000 req/hour) - ChangePasswordAPIView: Password change operations (1000 req/hour) - DeviceTokenAPIView: Device token registration (1000 req/hour) - DeviceTokenDeleteAPIView: Device token deletion (1000 req/hour) Changes: - Added UserRateThrottle import from rest_framework.throttling - Applied throttle_classes = [UserRateThrottle] to all 4 endpoints - Inherits default rate of 1000 requests/hour per user from base settings Security impact: - Prevents brute force attacks on password changes - Mitigates enumeration attacks via profile endpoint - Protects against DoS via excessive device token registration - Complements existing login/register rate limiting 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
533 lines
17 KiB
Python
533 lines
17 KiB
Python
"""
|
|
User views for KeepItGoing.
|
|
"""
|
|
|
|
from django.contrib.auth import get_user_model, login, logout
|
|
from django.shortcuts import render, redirect
|
|
from django.utils import timezone
|
|
from django.views import View
|
|
from rest_framework import generics, permissions, status
|
|
from rest_framework.response import Response
|
|
from rest_framework.throttling import UserRateThrottle
|
|
from rest_framework.views import APIView
|
|
from rest_framework.exceptions import AuthenticationFailed
|
|
from rest_framework_simplejwt.tokens import RefreshToken
|
|
from rest_framework_simplejwt.views import TokenObtainPairView as BaseTokenObtainPairView
|
|
from rest_framework_simplejwt.serializers import TokenObtainPairSerializer
|
|
|
|
from .serializers import (
|
|
UserSerializer,
|
|
UserRegistrationSerializer,
|
|
ChangePasswordSerializer,
|
|
DeviceTokenSerializer,
|
|
)
|
|
from .models import DeviceToken, EmailVerificationToken
|
|
from .throttles import LoginRateThrottle, RegisterRateThrottle
|
|
|
|
User = get_user_model()
|
|
|
|
|
|
# =============================================================================
|
|
# API Views
|
|
# =============================================================================
|
|
|
|
class RegisterAPIView(generics.CreateAPIView):
|
|
"""API endpoint for user registration."""
|
|
|
|
queryset = User.objects.all()
|
|
serializer_class = UserRegistrationSerializer
|
|
permission_classes = [permissions.AllowAny]
|
|
throttle_classes = [RegisterRateThrottle]
|
|
|
|
def create(self, request, *args, **kwargs):
|
|
serializer = self.get_serializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
user = serializer.save()
|
|
|
|
# Send verification email
|
|
from .utils import send_verification_email
|
|
send_verification_email(user, request)
|
|
|
|
# DO NOT generate tokens - user must verify email first
|
|
return Response({
|
|
'message': 'Registration successful! Please check your email to verify your account.',
|
|
'email': user.email,
|
|
'email_verified': False,
|
|
'is_approved': False,
|
|
}, status=status.HTTP_201_CREATED)
|
|
|
|
|
|
class UserProfileAPIView(generics.RetrieveUpdateAPIView):
|
|
"""API endpoint for user profile."""
|
|
|
|
serializer_class = UserSerializer
|
|
permission_classes = [permissions.IsAuthenticated]
|
|
throttle_classes = [UserRateThrottle]
|
|
|
|
def get_object(self):
|
|
return self.request.user
|
|
|
|
|
|
class ChangePasswordAPIView(APIView):
|
|
"""API endpoint for changing password."""
|
|
|
|
permission_classes = [permissions.IsAuthenticated]
|
|
throttle_classes = [UserRateThrottle]
|
|
|
|
def post(self, request):
|
|
serializer = ChangePasswordSerializer(
|
|
data=request.data,
|
|
context={'request': request}
|
|
)
|
|
serializer.is_valid(raise_exception=True)
|
|
|
|
request.user.set_password(serializer.validated_data['new_password'])
|
|
request.user.save()
|
|
|
|
return Response({'message': 'Password changed successfully.'})
|
|
|
|
|
|
class DeviceTokenAPIView(generics.ListCreateAPIView):
|
|
"""API endpoint for managing device tokens."""
|
|
|
|
serializer_class = DeviceTokenSerializer
|
|
permission_classes = [permissions.IsAuthenticated]
|
|
throttle_classes = [UserRateThrottle]
|
|
|
|
def get_queryset(self):
|
|
return DeviceToken.objects.filter(user=self.request.user)
|
|
|
|
|
|
class DeviceTokenDeleteAPIView(generics.DestroyAPIView):
|
|
"""API endpoint for deleting a device token."""
|
|
|
|
serializer_class = DeviceTokenSerializer
|
|
permission_classes = [permissions.IsAuthenticated]
|
|
throttle_classes = [UserRateThrottle]
|
|
|
|
def get_queryset(self):
|
|
return DeviceToken.objects.filter(user=self.request.user)
|
|
|
|
|
|
class VerifyEmailAPIView(APIView):
|
|
"""API endpoint for email verification."""
|
|
|
|
permission_classes = [permissions.AllowAny]
|
|
|
|
def post(self, request):
|
|
token_value = request.data.get('token')
|
|
|
|
if not token_value:
|
|
return Response(
|
|
{'error': 'Token is required.'},
|
|
status=status.HTTP_400_BAD_REQUEST
|
|
)
|
|
|
|
try:
|
|
token = EmailVerificationToken.objects.get(token=token_value)
|
|
except EmailVerificationToken.DoesNotExist:
|
|
return Response(
|
|
{'error': 'Invalid verification token.'},
|
|
status=status.HTTP_400_BAD_REQUEST
|
|
)
|
|
|
|
if token.is_used:
|
|
return Response(
|
|
{'error': 'This verification link has already been used.'},
|
|
status=status.HTTP_400_BAD_REQUEST
|
|
)
|
|
|
|
if token.is_expired():
|
|
return Response(
|
|
{'error': 'This verification link has expired. Please request a new one.'},
|
|
status=status.HTTP_400_BAD_REQUEST
|
|
)
|
|
|
|
# Mark email as verified
|
|
user = token.user
|
|
user.email_verified = True
|
|
user.email_verified_at = timezone.now()
|
|
user.save(update_fields=['email_verified', 'email_verified_at'])
|
|
|
|
# Mark token as used
|
|
token.is_used = True
|
|
token.used_at = timezone.now()
|
|
token.save(update_fields=['is_used', 'used_at'])
|
|
|
|
# Notify admins if not yet approved
|
|
if not user.is_approved:
|
|
from .utils import notify_admins_new_user
|
|
notify_admins_new_user(user)
|
|
|
|
return Response({
|
|
'message': 'Email verified successfully.',
|
|
'email_verified': True,
|
|
'is_approved': user.is_approved,
|
|
})
|
|
|
|
|
|
class ResendVerificationEmailAPIView(APIView):
|
|
"""API endpoint to resend verification email."""
|
|
|
|
permission_classes = [permissions.AllowAny]
|
|
|
|
def post(self, request):
|
|
email = request.data.get('email')
|
|
|
|
if not email:
|
|
return Response(
|
|
{'error': 'Email is required.'},
|
|
status=status.HTTP_400_BAD_REQUEST
|
|
)
|
|
|
|
try:
|
|
user = User.objects.get(email=email)
|
|
except User.DoesNotExist:
|
|
# Don't reveal if email exists
|
|
return Response({
|
|
'message': 'If that email is registered, a verification link has been sent.'
|
|
})
|
|
|
|
if user.email_verified:
|
|
return Response(
|
|
{'error': 'Email is already verified.'},
|
|
status=status.HTTP_400_BAD_REQUEST
|
|
)
|
|
|
|
from .utils import send_verification_email
|
|
send_verification_email(user, request)
|
|
|
|
return Response({
|
|
'message': 'Verification email sent.'
|
|
})
|
|
|
|
|
|
class CustomTokenObtainPairSerializer(TokenObtainPairSerializer):
|
|
"""Custom JWT token serializer that checks verification and approval."""
|
|
|
|
def validate(self, attrs):
|
|
# Authenticate user
|
|
data = super().validate(attrs)
|
|
|
|
# Check email verification
|
|
if not self.user.email_verified:
|
|
raise AuthenticationFailed({
|
|
'error': 'email_not_verified',
|
|
'message': 'Please verify your email address before logging in.',
|
|
'email': self.user.email,
|
|
})
|
|
|
|
# Check admin approval
|
|
if not self.user.is_approved:
|
|
raise AuthenticationFailed({
|
|
'error': 'not_approved',
|
|
'message': 'Your account is pending admin approval.',
|
|
'email': self.user.email,
|
|
})
|
|
|
|
return data
|
|
|
|
|
|
class TokenObtainPairView(BaseTokenObtainPairView):
|
|
"""Custom JWT token view using our custom serializer."""
|
|
|
|
serializer_class = CustomTokenObtainPairSerializer
|
|
throttle_classes = [LoginRateThrottle]
|
|
|
|
|
|
# =============================================================================
|
|
# Web Views (Django Templates)
|
|
# =============================================================================
|
|
|
|
class LoginView(View):
|
|
"""Web login page."""
|
|
|
|
def get(self, request):
|
|
if request.user.is_authenticated:
|
|
return redirect('dashboard')
|
|
|
|
# Check for special error conditions from authentication backend
|
|
error = None
|
|
email = None
|
|
can_resend = False
|
|
|
|
if 'login_error' in request.session:
|
|
error_type = request.session.pop('login_error')
|
|
email = request.session.pop('login_email', None)
|
|
|
|
if error_type == 'email_not_verified':
|
|
error = 'Please verify your email address before logging in.'
|
|
can_resend = True
|
|
elif error_type == 'not_approved':
|
|
error = 'Your account is pending admin approval.'
|
|
|
|
return render(request, 'users/login.html', {
|
|
'error': error,
|
|
'email': email,
|
|
'can_resend': can_resend,
|
|
})
|
|
|
|
def post(self, request):
|
|
from django.contrib.auth import authenticate
|
|
|
|
email = request.POST.get('email')
|
|
password = request.POST.get('password')
|
|
|
|
user = authenticate(request, username=email, password=password)
|
|
|
|
if user is not None:
|
|
login(request, user)
|
|
next_url = request.GET.get('next', 'dashboard')
|
|
return redirect(next_url)
|
|
|
|
# Check if error info was set by authentication backend
|
|
error = 'Invalid email or password.'
|
|
can_resend = False
|
|
|
|
if 'login_error' in request.session:
|
|
error_type = request.session.pop('login_error')
|
|
email_from_session = request.session.pop('login_email', None)
|
|
|
|
if error_type == 'email_not_verified':
|
|
error = 'Please verify your email address before logging in.'
|
|
can_resend = True
|
|
email = email_from_session
|
|
elif error_type == 'not_approved':
|
|
error = 'Your account is pending admin approval.'
|
|
email = email_from_session
|
|
|
|
return render(request, 'users/login.html', {
|
|
'error': error,
|
|
'email': email,
|
|
'can_resend': can_resend,
|
|
})
|
|
|
|
|
|
class LogoutView(View):
|
|
"""Web logout."""
|
|
|
|
def get(self, request):
|
|
logout(request)
|
|
return redirect('login')
|
|
|
|
def post(self, request):
|
|
logout(request)
|
|
return redirect('login')
|
|
|
|
|
|
class RegisterView(View):
|
|
"""Web registration page."""
|
|
|
|
def get(self, request):
|
|
if request.user.is_authenticated:
|
|
return redirect('dashboard')
|
|
return render(request, 'users/register.html')
|
|
|
|
def post(self, request):
|
|
email = request.POST.get('email')
|
|
username = request.POST.get('username')
|
|
password = request.POST.get('password')
|
|
password_confirm = request.POST.get('password_confirm')
|
|
first_name = request.POST.get('first_name', '')
|
|
last_name = request.POST.get('last_name', '')
|
|
timezone = request.POST.get('timezone', 'UTC')
|
|
|
|
errors = {}
|
|
|
|
if password != password_confirm:
|
|
errors['password_confirm'] = 'Passwords do not match.'
|
|
|
|
if User.objects.filter(email=email).exists():
|
|
errors['email'] = 'Email already registered.'
|
|
|
|
if User.objects.filter(username=username).exists():
|
|
errors['username'] = 'Username already taken.'
|
|
|
|
if errors:
|
|
return render(request, 'users/register.html', {
|
|
'errors': errors,
|
|
'email': email,
|
|
'username': username,
|
|
'first_name': first_name,
|
|
'last_name': last_name,
|
|
'timezone': timezone,
|
|
})
|
|
|
|
user = User.objects.create_user(
|
|
email=email,
|
|
username=username,
|
|
password=password,
|
|
first_name=first_name,
|
|
last_name=last_name,
|
|
timezone=timezone
|
|
)
|
|
|
|
# Send verification email
|
|
from .utils import send_verification_email
|
|
send_verification_email(user, request)
|
|
|
|
# DO NOT auto-login
|
|
return render(request, 'users/register_success.html', {
|
|
'email': user.email
|
|
})
|
|
|
|
|
|
class ProfileView(View):
|
|
"""Web profile page."""
|
|
|
|
def get(self, request):
|
|
if not request.user.is_authenticated:
|
|
return redirect('login')
|
|
return render(request, 'users/profile.html')
|
|
|
|
def post(self, request):
|
|
if not request.user.is_authenticated:
|
|
return redirect('login')
|
|
|
|
user = request.user
|
|
user.first_name = request.POST.get('first_name', '')
|
|
user.last_name = request.POST.get('last_name', '')
|
|
user.timezone = request.POST.get('timezone', 'UTC')
|
|
user.default_reminder_minutes = int(request.POST.get('default_reminder_minutes', 30))
|
|
user.email_notifications = request.POST.get('email_notifications') == 'on'
|
|
user.push_notifications = request.POST.get('push_notifications') == 'on'
|
|
user.save()
|
|
|
|
return render(request, 'users/profile.html', {
|
|
'success': 'Profile updated successfully.'
|
|
})
|
|
|
|
|
|
class VerifyEmailWebView(View):
|
|
"""Web view for email verification via link."""
|
|
|
|
def get(self, request, token):
|
|
try:
|
|
token_obj = EmailVerificationToken.objects.get(token=token)
|
|
except EmailVerificationToken.DoesNotExist:
|
|
return render(request, 'users/verify_email.html', {
|
|
'success': False,
|
|
'error': 'Invalid verification link.'
|
|
})
|
|
|
|
if token_obj.is_used:
|
|
return render(request, 'users/verify_email.html', {
|
|
'success': False,
|
|
'error': 'This verification link has already been used.'
|
|
})
|
|
|
|
if token_obj.is_expired():
|
|
return render(request, 'users/verify_email.html', {
|
|
'success': False,
|
|
'error': 'This verification link has expired.',
|
|
'can_resend': True,
|
|
'email': token_obj.user.email,
|
|
})
|
|
|
|
# Verify email
|
|
user = token_obj.user
|
|
user.email_verified = True
|
|
user.email_verified_at = timezone.now()
|
|
user.save(update_fields=['email_verified', 'email_verified_at'])
|
|
|
|
token_obj.is_used = True
|
|
token_obj.used_at = timezone.now()
|
|
token_obj.save(update_fields=['is_used', 'used_at'])
|
|
|
|
# Notify admins
|
|
if not user.is_approved:
|
|
from .utils import notify_admins_new_user
|
|
notify_admins_new_user(user)
|
|
|
|
return render(request, 'users/verify_email.html', {
|
|
'success': True,
|
|
'is_approved': user.is_approved,
|
|
})
|
|
|
|
|
|
class ResendVerificationWebView(View):
|
|
"""Web view for resending verification email."""
|
|
|
|
def get(self, request):
|
|
return render(request, 'users/resend_verification.html')
|
|
|
|
def post(self, request):
|
|
email = request.POST.get('email')
|
|
|
|
if not email:
|
|
return render(request, 'users/resend_verification.html', {
|
|
'error': 'Email is required.'
|
|
})
|
|
|
|
try:
|
|
user = User.objects.get(email=email)
|
|
except User.DoesNotExist:
|
|
# Don't reveal if email exists
|
|
return render(request, 'users/resend_verification.html', {
|
|
'success': True,
|
|
'message': 'If that email is registered, a verification link has been sent.'
|
|
})
|
|
|
|
if user.email_verified:
|
|
return render(request, 'users/resend_verification.html', {
|
|
'error': 'Email is already verified.'
|
|
})
|
|
|
|
from .utils import send_verification_email
|
|
send_verification_email(user, request)
|
|
|
|
return render(request, 'users/resend_verification.html', {
|
|
'success': True,
|
|
'message': 'Verification email sent. Please check your inbox.'
|
|
})
|
|
|
|
|
|
class ChangePasswordWebView(View):
|
|
"""Web view for changing password."""
|
|
|
|
def get(self, request):
|
|
if not request.user.is_authenticated:
|
|
return redirect('login')
|
|
return render(request, 'users/change_password.html')
|
|
|
|
def post(self, request):
|
|
if not request.user.is_authenticated:
|
|
return redirect('login')
|
|
|
|
old_password = request.POST.get('old_password')
|
|
new_password = request.POST.get('new_password')
|
|
confirm_password = request.POST.get('confirm_password')
|
|
|
|
errors = {}
|
|
|
|
# Validate old password
|
|
if not request.user.check_password(old_password):
|
|
errors['old_password'] = 'Current password is incorrect.'
|
|
|
|
# Validate new passwords match
|
|
if new_password != confirm_password:
|
|
errors['confirm_password'] = 'New passwords do not match.'
|
|
|
|
# Validate password length
|
|
if len(new_password) < 8:
|
|
errors['new_password'] = 'Password must be at least 8 characters long.'
|
|
|
|
# Validate new password is different from old
|
|
if old_password == new_password:
|
|
errors['new_password'] = 'New password must be different from current password.'
|
|
|
|
if errors:
|
|
return render(request, 'users/change_password.html', {'errors': errors})
|
|
|
|
# Change the password
|
|
request.user.set_password(new_password)
|
|
request.user.save()
|
|
|
|
# Re-login the user so their session isn't invalidated
|
|
from django.contrib.auth import update_session_auth_hash
|
|
update_session_auth_hash(request, request.user)
|
|
|
|
return render(request, 'users/change_password.html', {
|
|
'success': 'Password changed successfully!'
|
|
})
|