TIANHE
445e7ac290
- Fix token missing issue after email code registration/login - Add last login time update for code-based login - Support email login in addition to username login - Fix password login for code-registered users (allow setting password) - Fix referral code parameter passing from URL hash - Add email column to user management table - Improve profile page layout (align card heights, reorganize layout) - Add i18n support for Register Bonus, Referral Bonus, Code Lock Minutes, Code Max Attempts - Fix billing service add_credits to support reference_id parameter - Update password handling documentation
394 lines
15 KiB
Python
394 lines
15 KiB
Python
"""
|
|
Security Service - Handles Turnstile verification, rate limiting, and brute-force protection.
|
|
"""
|
|
import os
|
|
import json
|
|
import requests
|
|
from datetime import datetime, timedelta
|
|
from typing import Tuple, Optional, Dict, Any
|
|
from app.utils.db import get_db_connection
|
|
from app.utils.logger import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
# Singleton instance
|
|
_security_service = None
|
|
|
|
|
|
def get_security_service():
|
|
"""Get singleton SecurityService instance"""
|
|
global _security_service
|
|
if _security_service is None:
|
|
_security_service = SecurityService()
|
|
return _security_service
|
|
|
|
|
|
class SecurityService:
|
|
"""Security service for authentication protection"""
|
|
|
|
def __init__(self):
|
|
self._load_config()
|
|
|
|
def _load_config(self):
|
|
"""Load security configuration from environment variables"""
|
|
# Turnstile config
|
|
self.turnstile_site_key = os.getenv('TURNSTILE_SITE_KEY', '')
|
|
self.turnstile_secret_key = os.getenv('TURNSTILE_SECRET_KEY', '')
|
|
self.turnstile_enabled = bool(self.turnstile_site_key and self.turnstile_secret_key)
|
|
|
|
# IP rate limit config
|
|
self.ip_max_attempts = int(os.getenv('SECURITY_IP_MAX_ATTEMPTS', '10'))
|
|
self.ip_window_minutes = int(os.getenv('SECURITY_IP_WINDOW_MINUTES', '5'))
|
|
self.ip_block_minutes = int(os.getenv('SECURITY_IP_BLOCK_MINUTES', '15'))
|
|
|
|
# Account rate limit config
|
|
self.account_max_attempts = int(os.getenv('SECURITY_ACCOUNT_MAX_ATTEMPTS', '5'))
|
|
self.account_window_minutes = int(os.getenv('SECURITY_ACCOUNT_WINDOW_MINUTES', '60'))
|
|
self.account_block_minutes = int(os.getenv('SECURITY_ACCOUNT_BLOCK_MINUTES', '30'))
|
|
|
|
# Verification code rate limit
|
|
self.code_rate_limit_seconds = int(os.getenv('VERIFICATION_CODE_RATE_LIMIT', '60'))
|
|
self.code_ip_hourly_limit = int(os.getenv('VERIFICATION_CODE_IP_HOURLY_LIMIT', '10'))
|
|
|
|
def get_security_config(self) -> Dict[str, Any]:
|
|
"""Get public security config for frontend"""
|
|
return {
|
|
'turnstile_enabled': self.turnstile_enabled,
|
|
'turnstile_site_key': self.turnstile_site_key,
|
|
'registration_enabled': os.getenv('ENABLE_REGISTRATION', 'true').lower() == 'true',
|
|
'oauth_google_enabled': bool(os.getenv('GOOGLE_CLIENT_ID', '')),
|
|
'oauth_github_enabled': bool(os.getenv('GITHUB_CLIENT_ID', '')),
|
|
}
|
|
|
|
# =========================================================================
|
|
# Turnstile Verification
|
|
# =========================================================================
|
|
|
|
def verify_turnstile(self, token: str, ip_address: str = None) -> Tuple[bool, str]:
|
|
"""
|
|
Verify Cloudflare Turnstile token.
|
|
|
|
Returns:
|
|
(success, message)
|
|
"""
|
|
if not self.turnstile_enabled:
|
|
# If Turnstile is not configured, skip verification
|
|
return True, 'turnstile_disabled'
|
|
|
|
if not token:
|
|
return False, 'Missing Turnstile token'
|
|
|
|
try:
|
|
response = requests.post(
|
|
'https://challenges.cloudflare.com/turnstile/v0/siteverify',
|
|
data={
|
|
'secret': self.turnstile_secret_key,
|
|
'response': token,
|
|
'remoteip': ip_address
|
|
},
|
|
timeout=10
|
|
)
|
|
result = response.json()
|
|
|
|
if result.get('success'):
|
|
return True, 'verified'
|
|
else:
|
|
error_codes = result.get('error-codes', [])
|
|
logger.warning(f"Turnstile verification failed: {error_codes}")
|
|
return False, 'Turnstile verification failed'
|
|
|
|
except requests.RequestException as e:
|
|
logger.error(f"Turnstile API error: {e}")
|
|
# On API error, we might want to allow (fail-open) or deny (fail-closed)
|
|
# For security, we'll deny
|
|
return False, 'Turnstile service unavailable'
|
|
|
|
# =========================================================================
|
|
# Rate Limiting & Brute-Force Protection
|
|
# =========================================================================
|
|
|
|
def record_login_attempt(self, identifier: str, identifier_type: str,
|
|
success: bool, ip_address: str = None,
|
|
user_agent: str = None) -> bool:
|
|
"""
|
|
Record a login attempt for rate limiting.
|
|
|
|
Args:
|
|
identifier: IP address or username
|
|
identifier_type: 'ip' or 'account'
|
|
success: Whether the attempt was successful
|
|
ip_address: Client IP address
|
|
user_agent: Client user agent string
|
|
"""
|
|
try:
|
|
with get_db_connection() as db:
|
|
cur = db.cursor()
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO qd_login_attempts
|
|
(identifier, identifier_type, success, ip_address, user_agent)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
""",
|
|
(identifier, identifier_type, success, ip_address, user_agent)
|
|
)
|
|
db.commit()
|
|
cur.close()
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Failed to record login attempt: {e}")
|
|
return False
|
|
|
|
def is_blocked(self, identifier: str, identifier_type: str) -> Tuple[bool, int]:
|
|
"""
|
|
Check if an identifier (IP or account) is blocked due to too many failed attempts.
|
|
|
|
Returns:
|
|
(is_blocked, remaining_seconds)
|
|
"""
|
|
try:
|
|
if identifier_type == 'ip':
|
|
max_attempts = self.ip_max_attempts
|
|
window_minutes = self.ip_window_minutes
|
|
block_minutes = self.ip_block_minutes
|
|
else: # account
|
|
max_attempts = self.account_max_attempts
|
|
window_minutes = self.account_window_minutes
|
|
block_minutes = self.account_block_minutes
|
|
|
|
with get_db_connection() as db:
|
|
cur = db.cursor()
|
|
|
|
# Count failed attempts in the time window
|
|
window_start = datetime.now() - timedelta(minutes=window_minutes)
|
|
cur.execute(
|
|
"""
|
|
SELECT COUNT(*) as count, MAX(attempt_time) as last_attempt
|
|
FROM qd_login_attempts
|
|
WHERE identifier = ? AND identifier_type = ?
|
|
AND success = FALSE AND attempt_time > ?
|
|
""",
|
|
(identifier, identifier_type, window_start)
|
|
)
|
|
row = cur.fetchone()
|
|
cur.close()
|
|
|
|
if not row:
|
|
return False, 0
|
|
|
|
failed_count = row['count'] or 0
|
|
last_attempt = row['last_attempt']
|
|
|
|
if failed_count >= max_attempts:
|
|
# Check if still in block period
|
|
if last_attempt:
|
|
block_until = last_attempt + timedelta(minutes=block_minutes)
|
|
if datetime.now() < block_until:
|
|
remaining = int((block_until - datetime.now()).total_seconds())
|
|
return True, remaining
|
|
|
|
return False, 0
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to check block status: {e}")
|
|
return False, 0
|
|
|
|
def check_login_allowed(self, username: str, ip_address: str) -> Tuple[bool, str]:
|
|
"""
|
|
Check if login is allowed for the given username and IP.
|
|
|
|
Returns:
|
|
(allowed, message)
|
|
"""
|
|
# Check IP block
|
|
ip_blocked, ip_remaining = self.is_blocked(ip_address, 'ip')
|
|
if ip_blocked:
|
|
minutes = ip_remaining // 60
|
|
return False, f'Too many failed attempts from this IP. Try again in {minutes + 1} minutes.'
|
|
|
|
# Check account block
|
|
account_blocked, account_remaining = self.is_blocked(username, 'account')
|
|
if account_blocked:
|
|
minutes = account_remaining // 60
|
|
return False, f'Account temporarily locked due to too many failed attempts. Try again in {minutes + 1} minutes.'
|
|
|
|
return True, 'allowed'
|
|
|
|
def clear_login_attempts(self, identifier: str, identifier_type: str) -> bool:
|
|
"""
|
|
Clear login attempts for an identifier (called after successful login).
|
|
"""
|
|
try:
|
|
with get_db_connection() as db:
|
|
cur = db.cursor()
|
|
cur.execute(
|
|
"""
|
|
DELETE FROM qd_login_attempts
|
|
WHERE identifier = ? AND identifier_type = ?
|
|
""",
|
|
(identifier, identifier_type)
|
|
)
|
|
db.commit()
|
|
cur.close()
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Failed to clear login attempts: {e}")
|
|
return False
|
|
|
|
# =========================================================================
|
|
# Security Audit Logging
|
|
# =========================================================================
|
|
|
|
def log_security_event(self, action: str, user_id: int = None,
|
|
ip_address: str = None, user_agent: str = None,
|
|
details: dict = None) -> bool:
|
|
"""
|
|
Log a security-related event.
|
|
|
|
Args:
|
|
action: Event type (login, logout, register, reset_password, etc.)
|
|
user_id: User ID if applicable
|
|
ip_address: Client IP
|
|
user_agent: Client user agent
|
|
details: Additional details as dict
|
|
"""
|
|
try:
|
|
details_json = json.dumps(details) if details else None
|
|
|
|
with get_db_connection() as db:
|
|
cur = db.cursor()
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO qd_security_logs
|
|
(user_id, action, ip_address, user_agent, details)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
""",
|
|
(user_id, action, ip_address, user_agent, details_json)
|
|
)
|
|
db.commit()
|
|
cur.close()
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Failed to log security event: {e}")
|
|
return False
|
|
|
|
# =========================================================================
|
|
# Verification Code Rate Limiting
|
|
# =========================================================================
|
|
|
|
def can_send_verification_code(self, email: str, ip_address: str) -> Tuple[bool, str]:
|
|
"""
|
|
Check if we can send a verification code to this email from this IP.
|
|
|
|
Returns:
|
|
(allowed, message)
|
|
"""
|
|
try:
|
|
with get_db_connection() as db:
|
|
cur = db.cursor()
|
|
|
|
# Check email rate limit (one code per minute per email)
|
|
rate_limit_time = datetime.now() - timedelta(seconds=self.code_rate_limit_seconds)
|
|
cur.execute(
|
|
"""
|
|
SELECT COUNT(*) as count FROM qd_verification_codes
|
|
WHERE email = ? AND created_at > ?
|
|
""",
|
|
(email, rate_limit_time)
|
|
)
|
|
row = cur.fetchone()
|
|
if row and row['count'] > 0:
|
|
return False, f'Please wait {self.code_rate_limit_seconds} seconds before requesting another code'
|
|
|
|
# Check IP hourly limit
|
|
hour_ago = datetime.now() - timedelta(hours=1)
|
|
cur.execute(
|
|
"""
|
|
SELECT COUNT(*) as count FROM qd_verification_codes
|
|
WHERE ip_address = ? AND created_at > ?
|
|
""",
|
|
(ip_address, hour_ago)
|
|
)
|
|
row = cur.fetchone()
|
|
if row and row['count'] >= self.code_ip_hourly_limit:
|
|
return False, 'Too many verification code requests from this IP. Try again later.'
|
|
|
|
cur.close()
|
|
return True, 'allowed'
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to check verification code rate limit: {e}")
|
|
return True, 'allowed' # Fail open on DB errors
|
|
|
|
# =========================================================================
|
|
# Password Strength Validation
|
|
# =========================================================================
|
|
|
|
def validate_password_strength(self, password: str) -> Tuple[bool, str]:
|
|
"""
|
|
Validate password meets minimum security requirements.
|
|
|
|
Requirements:
|
|
- At least 8 characters
|
|
- Contains at least one uppercase letter
|
|
- Contains at least one lowercase letter
|
|
- Contains at least one digit
|
|
|
|
Returns:
|
|
(valid, message)
|
|
"""
|
|
if len(password) < 8:
|
|
return False, 'Password must be at least 8 characters long'
|
|
|
|
if not any(c.isupper() for c in password):
|
|
return False, 'Password must contain at least one uppercase letter'
|
|
|
|
if not any(c.islower() for c in password):
|
|
return False, 'Password must contain at least one lowercase letter'
|
|
|
|
if not any(c.isdigit() for c in password):
|
|
return False, 'Password must contain at least one digit'
|
|
|
|
return True, 'valid'
|
|
|
|
# =========================================================================
|
|
# Cleanup
|
|
# =========================================================================
|
|
|
|
def cleanup_old_records(self, days: int = 7) -> int:
|
|
"""
|
|
Clean up old login attempts and expired verification codes.
|
|
|
|
Returns:
|
|
Number of records deleted
|
|
"""
|
|
deleted = 0
|
|
cutoff = datetime.now() - timedelta(days=days)
|
|
|
|
try:
|
|
with get_db_connection() as db:
|
|
cur = db.cursor()
|
|
|
|
# Clean old login attempts
|
|
cur.execute(
|
|
"DELETE FROM qd_login_attempts WHERE attempt_time < ?",
|
|
(cutoff,)
|
|
)
|
|
deleted += cur.rowcount or 0
|
|
|
|
# Clean expired verification codes
|
|
cur.execute(
|
|
"DELETE FROM qd_verification_codes WHERE expires_at < ?",
|
|
(cutoff,)
|
|
)
|
|
deleted += cur.rowcount or 0
|
|
|
|
db.commit()
|
|
cur.close()
|
|
|
|
logger.info(f"Security cleanup: deleted {deleted} old records")
|
|
return deleted
|
|
|
|
except Exception as e:
|
|
logger.error(f"Security cleanup failed: {e}")
|
|
return 0
|