huge updates

This commit is contained in:
sankalp committed 2025-04-22 03:47:31 +05:45
1 parent 9b721e10d4
commit a9a5f1bc10
40 files changed
+6878 -2378

No files matched your search

+267 -49
View File
@@ -2,21 +2,161 @@
import base64
import random
import os
import magic
# Core Django imports
from django.core.mail import send_mail
from django.template.loader import render_to_string
from django.utils import timezone
from django.core.files.storage import default_storage
from django.core.files.base import ContentFile
from django.conf import settings
import logging
from django.core.exceptions import ValidationError
# Third-party app imports
import pyotp
# Local app imports
from app.models import ResetToken, Staff, Student, TOTPSecret
from app.models import ResetToken, Staff, Student, TOTPSecret, OTPAttempt
# OTP expiration time in seconds (5 minutes)
OTP_EXPIRY = int(os.getenv('OTP_EXPIRY', 300))
logger = logging.getLogger(__name__)
class FileUploadError(Exception):
"""Custom exception for file upload errors"""
pass
def validate_file_type(file, allowed_types):
"""
Validate file type using python-magic
Args:
file: Uploaded file object
allowed_types: List of allowed MIME types
Returns:
bool: True if file type is allowed, False otherwise
"""
try:
# Read first 2048 bytes to determine file type
file.seek(0)
file_content = file.read(2048)
file.seek(0) # Reset file pointer
mime_type = magic.from_buffer(file_content, mime=True)
return mime_type in allowed_types
except Exception as e:
logger.error(f"Error validating file type: {e}")
return False
def validate_file_size(file, max_size_bytes):
"""
Validate file size
Args:
file: Uploaded file object
max_size_bytes: Maximum allowed file size in bytes
Returns:
bool: True if file size is within limit, False otherwise
"""
try:
return file.size <= max_size_bytes
except Exception as e:
logger.error(f"Error validating file size: {e}")
return False
def sanitize_filename(filename):
"""
Sanitize filename to prevent directory traversal and other security issues
Args:
filename: Original filename
Returns:
str: Sanitized filename
"""
# Remove any path components
filename = os.path.basename(filename)
# Replace any non-alphanumeric characters (except ._-) with underscore
filename = ''.join(c if c.isalnum() or c in '._-' else '_' for c in filename)
return filename
def handle_file_upload(file, upload_path, allowed_types, max_size_bytes):
"""
Handle file upload with security checks
Args:
file: Uploaded file object
upload_path: Path where file should be uploaded
allowed_types: List of allowed MIME types
max_size_bytes: Maximum allowed file size in bytes
Returns:
str: Path to uploaded file if successful
Raises:
FileUploadError: If file validation fails
"""
try:
# Validate file type
if not validate_file_type(file, allowed_types):
raise FileUploadError(f"Invalid file type. Allowed types: {', '.join(allowed_types)}")
# Validate file size
if not validate_file_size(file, max_size_bytes):
raise FileUploadError(f"File size too large. Maximum size: {max_size_bytes/1024/1024}MB")
# Sanitize filename
filename = sanitize_filename(file.name)
# Create full upload path
full_path = os.path.join(upload_path, filename)
# Save file
path = default_storage.save(full_path, ContentFile(file.read()))
return path
except Exception as e:
logger.error(f"Error handling file upload: {e}")
raise FileUploadError(str(e))
def cleanup_failed_upload(file_path):
"""
Clean up failed file upload
Args:
file_path: Path to the file that needs to be cleaned up
"""
try:
if default_storage.exists(file_path):
default_storage.delete(file_path)
except Exception as e:
logger.error(f"Error cleaning up failed upload: {e}")
# Constants for file uploads
ALLOWED_DOCUMENT_TYPES = [
'application/pdf',
'application/msword',
'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
'application/vnd.ms-powerpoint',
'application/vnd.openxmlformats-officedocument.presentationml.presentation',
'text/plain',
'application/rtf'
]
ALLOWED_IMAGE_TYPES = [
'image/jpeg',
'image/png',
'image/gif'
]
MAX_DOCUMENT_SIZE = 10 * 1024 * 1024 # 10MB
MAX_IMAGE_SIZE = 5 * 1024 * 1024 # 5MB
# --------------------------------------------------------------------
# OTP and Secret Key Generation
@@ -52,18 +192,36 @@ def store_secret_key(identifier, secret_key):
def verify_otp(identifier, otp):
"""Verify OTP using TOTP"""
"""Verify OTP using TOTP with improved security"""
try:
# Check if identifier is locked out
attempt, _ = OTPAttempt.objects.get_or_create(identifier=identifier)
if attempt.is_locked_out():
logger.warning(f"Identifier {identifier} is locked out from OTP attempts")
return False
# Get the most recent secret for this identifier
secret = TOTPSecret.objects.filter(identifier=identifier).latest("created_at")
# Check if secret is expired
if secret.is_expired():
attempt.increment_attempts()
return False
# Verify OTP
totp = pyotp.TOTP(secret.secret_key, interval=OTP_EXPIRY)
return totp.verify(otp)
is_valid = totp.verify(otp)
if is_valid:
# Reset attempts on successful verification
attempt.reset_attempts()
# Clean up expired tokens
cleanup_expired_tokens()
else:
# Increment failed attempts
attempt.increment_attempts()
return is_valid
except TOTPSecret.DoesNotExist:
return False
@@ -73,44 +231,79 @@ def verify_otp(identifier, otp):
# --------------------------------------------------------------------
def validate_phone_number(phone):
"""Validate phone number format with improved validation"""
if not phone or not isinstance(phone, str):
raise ValidationError("Phone number must be a string")
# Remove any non-digit characters
phone = ''.join(filter(str.isdigit, phone))
# Validate length and format
if len(phone) < 10 or len(phone) > 15:
raise ValidationError("Phone number must be between 10 and 15 digits")
# Validate country code if present
if phone.startswith('0'):
raise ValidationError("Phone number cannot start with 0")
# Additional validation for specific countries if needed
if phone.startswith('91'): # India
if len(phone) != 12:
raise ValidationError("Indian phone numbers must be 12 digits including country code")
return True
def send_otp_sms(phone, otp):
"""Send OTP via SMS using a service"""
"""Send OTP via SMS with improved error handling"""
try:
# Get user details from phone number
# Validate phone number first
validate_phone_number(phone)
# Check if phone is locked out
attempt, _ = OTPAttempt.objects.get_or_create(identifier=phone)
if attempt.is_locked_out():
logger.warning(f"Phone {phone} is locked out from OTP attempts")
return False
# Get user name if available
user = None
student = Student.objects.filter(phone=phone).first()
staff = Staff.objects.filter(phone=phone).first()
user = student or staff
# Get user's name
user_name = user.name if user else "User"
# Calculate expiry time in minutes
expiry_minutes = OTP_EXPIRY // 60
# Render SMS message using template
message = render_to_string(
"login/sms_templates/otp_message.txt",
{"user_name": user_name, "otp": otp, "expiry_time": expiry_minutes},
)
# This is a placeholder. You need to replace with actual SMS API implementation
# Example using a generic SMS API
# api_url = "https://api.example.com/send"
# payload = {
# "apikey": settings.SMS_API_KEY,
# "to": phone,
# "message": message
# }
# response = requests.post(api_url, json=payload)
# return response.status_code == 200
print("\nSMS Message Preview:")
print("-" * 50)
print(message)
print("-" * 50)
if student:
user = student
elif staff:
user = staff
# Prepare context for template
context = {
'user_name': user.name if user else 'User',
'otp': otp,
'expiry_time': OTP_EXPIRY // 60 # Convert seconds to minutes
}
# Render message from template
message = render_to_string('login/sms_templates/otp_message.txt', context)
# Your SMS service implementation here
# Example using Twilio:
# from twilio.rest import Client
# client = Client(settings.TWILIO_ACCOUNT_SID, settings.TWILIO_AUTH_TOKEN)
# message = client.messages.create(
# body=message,
# from_=settings.TWILIO_PHONE_NUMBER,
# to=f"+{phone}"
# )
# For now, just log the OTP
logger.info(f"OTP message would be sent to {phone}: {message}")
return True
except ValidationError as e:
logger.error(f"Invalid phone number format: {str(e)}")
return False
except Exception as e:
print(f"Error sending OTP: {str(e)}")
logger.error(f"Failed to send SMS to {phone}: {str(e)}")
return False
@@ -176,19 +369,25 @@ def verify_reset_code(email, code):
def store_reset_token(token, identifier):
"""Store reset token in database"""
# Delete any existing tokens for this identifier
ResetToken.objects.filter(identifier=identifier).delete()
# Create new token with expiration (15 minutes)
expires_at = timezone.now() + timezone.timedelta(minutes=15)
return ResetToken.objects.create(
token=token, identifier=identifier, expires_at=expires_at
)
"""Store reset token with improved session management"""
try:
# Clean up any existing tokens for this identifier
ResetToken.objects.filter(identifier=identifier).delete()
# Create new token with expiration
expires_at = timezone.now() + timezone.timedelta(minutes=15) # 15 minutes expiration
return ResetToken.objects.create(
token=token,
identifier=identifier,
expires_at=expires_at
)
except Exception as e:
logger.error(f"Error storing reset token: {str(e)}")
return None
def verify_reset_token(token, identifier):
"""Verify reset token from database"""
"""Verify reset token with improved security"""
try:
reset_token = ResetToken.objects.get(token=token)
@@ -196,7 +395,14 @@ def verify_reset_token(token, identifier):
if reset_token.is_expired():
return False
return reset_token.identifier == identifier
# Verify identifier matches
if reset_token.identifier != identifier:
return False
# Clean up expired tokens
cleanup_expired_tokens()
return True
except ResetToken.DoesNotExist:
return False
@@ -207,6 +413,18 @@ def verify_reset_token(token, identifier):
def cleanup_expired_tokens():
"""Clean up expired TOTP secrets and reset tokens"""
TOTPSecret.objects.filter(expires_at__lt=timezone.now()).delete()
ResetToken.objects.filter(expires_at__lt=timezone.now()).delete()
"""Clean up expired OTP secrets and reset tokens"""
try:
# Delete expired OTP secrets
expired_otps = TOTPSecret.objects.filter(expires_at__lt=timezone.now())
expired_otps_count = expired_otps.count()
expired_otps.delete()
# Delete expired reset tokens
expired_reset_tokens = ResetToken.objects.filter(expires_at__lt=timezone.now())
expired_reset_count = expired_reset_tokens.count()
expired_reset_tokens.delete()
logger.info(f"Cleaned up {expired_otps_count} expired OTPs and {expired_reset_count} expired reset tokens")
except Exception as e:
logger.error(f"Error cleaning up expired tokens: {str(e)}")