huge updates
This commit is contained in:
57 files changed
+10143
-4185
No files matched your search
+107
-91
@@ -1,128 +1,119 @@
|
||||
import random
|
||||
import string
|
||||
from django.core.cache import cache
|
||||
import requests
|
||||
from firebase_admin import messaging, initialize_app, get_app, credentials
|
||||
import os
|
||||
from django.template.loader import render_to_string
|
||||
from django.core.mail import send_mail
|
||||
from django.conf import settings
|
||||
from django.contrib.auth.tokens import default_token_generator
|
||||
from django.utils.http import urlsafe_base64_encode
|
||||
from django.utils.encoding import force_bytes
|
||||
from django.utils import timezone
|
||||
import pyotp
|
||||
import base64
|
||||
from .models import TOTPSecret, ResetToken, Student, Staff
|
||||
|
||||
# OTP expiration time in seconds (5 minutes)
|
||||
OTP_EXPIRY = 300
|
||||
|
||||
|
||||
def generate_otp(length=6):
|
||||
"""Generate a numeric OTP of specified length"""
|
||||
digits = string.digits
|
||||
return "".join(random.choice(digits) for _ in range(length))
|
||||
def generate_secret_key():
|
||||
"""Generate a random secret key for TOTP"""
|
||||
return base64.b32encode(random.getrandbits(160).to_bytes(20, 'big')).decode('utf-8')
|
||||
|
||||
|
||||
def store_otp(phone, otp):
|
||||
"""Store OTP in cache with phone as key"""
|
||||
cache_key = f"otp_{phone}"
|
||||
cache.set(cache_key, otp, OTP_EXPIRY)
|
||||
return True
|
||||
def generate_otp(secret_key):
|
||||
"""Generate a TOTP using the secret key"""
|
||||
totp = pyotp.TOTP(secret_key, interval=OTP_EXPIRY)
|
||||
return totp.now()
|
||||
|
||||
|
||||
def verify_otp(phone, otp):
|
||||
"""Verify if the provided OTP matches the stored OTP for the phone"""
|
||||
cache_key = f"otp_{phone}"
|
||||
stored_otp = cache.get(cache_key)
|
||||
def store_secret_key(identifier, secret_key):
|
||||
"""Store secret key in database"""
|
||||
# Delete any existing secrets for this identifier
|
||||
TOTPSecret.objects.filter(identifier=identifier).delete()
|
||||
|
||||
# Create new secret with expiration
|
||||
expires_at = timezone.now() + timezone.timedelta(seconds=OTP_EXPIRY)
|
||||
return TOTPSecret.objects.create(
|
||||
identifier=identifier,
|
||||
secret_key=secret_key,
|
||||
expires_at=expires_at
|
||||
)
|
||||
|
||||
if stored_otp and stored_otp == otp:
|
||||
# Delete the OTP after successful verification
|
||||
cache.delete(cache_key)
|
||||
return True
|
||||
return False
|
||||
|
||||
def verify_otp(identifier, otp):
|
||||
"""Verify OTP using TOTP"""
|
||||
try:
|
||||
# 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():
|
||||
return False
|
||||
|
||||
# Verify OTP
|
||||
totp = pyotp.TOTP(secret.secret_key, interval=OTP_EXPIRY)
|
||||
return totp.verify(otp)
|
||||
except TOTPSecret.DoesNotExist:
|
||||
return False
|
||||
|
||||
|
||||
def send_otp_sms(phone, otp):
|
||||
"""Send OTP via SMS using a service like Twilio, MSG91, etc."""
|
||||
"""Send OTP via SMS using a service"""
|
||||
try:
|
||||
# Get user details from phone number
|
||||
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": f"Your OTP for SMS login is: {otp}. Valid for 5 minutes."
|
||||
# "message": message
|
||||
# }
|
||||
# response = requests.post(api_url, json=payload)
|
||||
# return response.status_code == 200
|
||||
|
||||
print(f"OTP for {phone}: {otp}")
|
||||
print("\nSMS Message Preview:")
|
||||
print("-" * 50)
|
||||
print(message)
|
||||
print("-" * 50)
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"Error sending OTP: {str(e)}")
|
||||
return False
|
||||
|
||||
|
||||
def send_push_notification(title, message, tokens):
|
||||
"""
|
||||
Send push notification using Firebase Cloud Messaging
|
||||
"""
|
||||
try:
|
||||
try:
|
||||
app = get_app()
|
||||
except ValueError:
|
||||
# Initialize Firebase Admin SDK if not already initialized
|
||||
cred_path = os.path.join(
|
||||
os.path.dirname(os.path.dirname(__file__)),
|
||||
"firebase-key.json",
|
||||
)
|
||||
if not os.path.exists(cred_path):
|
||||
print(f"Firebase credentials file not found at {cred_path}")
|
||||
return
|
||||
|
||||
cred = credentials.Certificate(cred_path)
|
||||
app = initialize_app(cred)
|
||||
|
||||
if not tokens:
|
||||
print("No FCM tokens available to send notifications")
|
||||
return
|
||||
|
||||
# Create message
|
||||
message = messaging.MulticastMessage(
|
||||
notification=messaging.Notification(
|
||||
title=title,
|
||||
body=message,
|
||||
),
|
||||
tokens=list(tokens),
|
||||
)
|
||||
|
||||
# Send message
|
||||
response = messaging.send_multicast(message)
|
||||
|
||||
if response.failure_count > 0:
|
||||
failures = []
|
||||
for idx, resp in enumerate(response.responses):
|
||||
if not resp.success:
|
||||
failures.append(
|
||||
f"Failed to send message to token {tokens[idx]}: {resp.exception}"
|
||||
)
|
||||
print("\n".join(failures))
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error sending notification: {e}")
|
||||
# Don't raise the exception to prevent save from failing
|
||||
pass
|
||||
|
||||
|
||||
def send_password_reset_email(user, request, email):
|
||||
"""
|
||||
Send password reset code to user's email using HTML template
|
||||
"""
|
||||
try:
|
||||
# Generate a reset code (OTP)
|
||||
reset_code = generate_otp(length=6)
|
||||
|
||||
# Store the reset code in cache with user's email as key
|
||||
cache_key = f"reset_code_{email}"
|
||||
cache.set(cache_key, reset_code, OTP_EXPIRY) # 5 minutes expiry
|
||||
# Generate a secret key for TOTP
|
||||
secret_key = generate_secret_key()
|
||||
|
||||
# Store the secret key in database
|
||||
store_secret_key(email, secret_key)
|
||||
|
||||
# Generate TOTP
|
||||
totp = pyotp.TOTP(secret_key, interval=OTP_EXPIRY)
|
||||
reset_code = totp.now()
|
||||
|
||||
# Build email context
|
||||
context = {
|
||||
@@ -156,15 +147,40 @@ def send_password_reset_email(user, request, email):
|
||||
|
||||
|
||||
def verify_reset_code(email, code):
|
||||
"""
|
||||
Verify if the provided reset code matches the stored code for the email
|
||||
"""
|
||||
cache_key = f"reset_code_{email}"
|
||||
stored_code = cache.get(cache_key)
|
||||
"""Verify password reset code using TOTP"""
|
||||
return verify_otp(email, code)
|
||||
|
||||
if stored_code and stored_code == code:
|
||||
# Delete the code after successful verification
|
||||
cache.delete(cache_key)
|
||||
return True
|
||||
return False
|
||||
|
||||
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
|
||||
)
|
||||
|
||||
|
||||
def verify_reset_token(token, identifier):
|
||||
"""Verify reset token from database"""
|
||||
try:
|
||||
reset_token = ResetToken.objects.get(token=token)
|
||||
|
||||
# Check if token is expired
|
||||
if reset_token.is_expired():
|
||||
return False
|
||||
|
||||
return reset_token.identifier == identifier
|
||||
except ResetToken.DoesNotExist:
|
||||
return False
|
||||
|
||||
|
||||
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()
|
||||
|
||||
Reference in new issue
Block a user