gavindiaz/DinQuant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: improve authentication and user management

Browse Source 
- 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
This commit is contained in:
TIANHE
2026-01-14 20:42:10 +08:00
parent fa721c034c
commit 445e7ac290
33 changed files with 7715 additions and 234 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend_api_python/app/routes/__init__.py
+1 -1
View File
@@ -25,7 +25,7 @@ def register_routes(app: Flask):
from app.routes.strategy_code import strategy_code_bp
app.register_blueprint(health_bp)
app.register_blueprint(auth_bp, url_prefix='/api/user')
app.register_blueprint(auth_bp, url_prefix='/api/auth') # Auth routes
app.register_blueprint(user_bp, url_prefix='/api/users') # User management
app.register_blueprint(kline_bp, url_prefix='/api/indicator')
app.register_blueprint(analysis_bp, url_prefix='/api/analysis')
backend_api_python/app/routes/analysis.py
+19
View File
@@ -140,6 +140,25 @@ def multi_analysis():
logger.info(f"Analyze request: {market}:{symbol}, use_multi_agent={use_multi_agent}, model={model}")
# Step 0: Check billing (计费检查)
from app.services.billing_service import get_billing_service
billing_success, billing_msg = get_billing_service().check_and_consume(
user_id=user_id,
feature='ai_analysis',
reference_id=f'{market}:{symbol}'
)
if not billing_success:
if 'insufficient_credits' in billing_msg:
parts = billing_msg.split(':')
current = parts[1] if len(parts) > 1 else '0'
required = parts[2] if len(parts) > 2 else '?'
return jsonify({
'code': 0,
'msg': f'Insufficient credits. Current: {current}, Required: {required}',
'data': {'error_type': 'insufficient_credits', 'current': current, 'required': required}
}), 402
return jsonify({'code': 0, 'msg': billing_msg, 'data': None}), 400
# Step 1: Create a "pending" task record first (so user can see progress in history)
task_id = _store_task(user_id, market, symbol, model or '', language, 'pending', result={}, error_message='')
backend_api_python/app/routes/auth.py
+838 -6
View File
@@ -1,11 +1,11 @@
"""
Authentication API Routes
Handles login, logout, and user info retrieval.
Handles login, logout, registration, password reset, and OAuth authentication.
Supports both multi-user (database) and single-user (legacy) modes.
"""
import os
from flask import Blueprint, request, jsonify, g
from flask import Blueprint, request, jsonify, g, redirect
from app.config.settings import Config
from app.utils.auth import generate_token, login_required, authenticate_legacy
from app.utils.logger import get_logger
@@ -20,6 +20,50 @@ def _is_single_user_mode() -> bool:
return os.getenv('SINGLE_USER_MODE', 'false').lower() == 'true'
def _get_client_ip() -> str:
"""Get client IP address from request"""
# Check for proxy headers
if request.headers.get('X-Forwarded-For'):
return request.headers.get('X-Forwarded-For').split(',')[0].strip()
if request.headers.get('X-Real-IP'):
return request.headers.get('X-Real-IP')
return request.remote_addr or '0.0.0.0'
def _get_user_agent() -> str:
"""Get user agent from request"""
return request.headers.get('User-Agent', '')[:500]
# =============================================================================
# Security Config Endpoint
# =============================================================================
@auth_bp.route('/security-config', methods=['GET'])
def get_security_config():
"""
Get public security configuration for frontend.
Returns:
turnstile_enabled: bool
turnstile_site_key: str
registration_enabled: bool
oauth_google_enabled: bool
oauth_github_enabled: bool
"""
try:
from app.services.security_service import get_security_service
config = get_security_service().get_security_config()
return jsonify({'code': 1, 'msg': 'success', 'data': config})
except Exception as e:
logger.error(f"get_security_config error: {e}")
return jsonify({'code': 0, 'msg': str(e), 'data': None}), 500
# =============================================================================
# Login Endpoint (Enhanced with security)
# =============================================================================
@auth_bp.route('/login', methods=['POST'])
def login():
"""
@@ -28,30 +72,62 @@ def login():
Request body:
username: str
password: str
turnstile_token: str (optional, required if Turnstile is enabled)
Returns:
token: JWT token
userinfo: User information
"""
ip_address = _get_client_ip()
user_agent = _get_user_agent()
try:
from app.services.security_service import get_security_service
security = get_security_service()
data = request.get_json()
if not data:
return jsonify({'code': 400, 'msg': 'No data provided', 'data': None}), 400
username = data.get('username') or data.get('account')
password = data.get('password')
turnstile_token = data.get('turnstile_token')
if not username or not password:
return jsonify({'code': 400, 'msg': 'Missing username or password', 'data': None}), 400
return jsonify({'code': 400, 'msg': 'Missing username/email or password', 'data': None}), 400
# Step 1: Verify Turnstile (if enabled)
turnstile_ok, turnstile_msg = security.verify_turnstile(turnstile_token, ip_address)
if not turnstile_ok:
return jsonify({'code': 0, 'msg': turnstile_msg, 'data': None}), 400
# Step 2: Check rate limiting
allowed, block_msg = security.check_login_allowed(username, ip_address)
if not allowed:
return jsonify({'code': 0, 'msg': block_msg, 'data': {'blocked': True}}), 429
is_demo = os.getenv('IS_DEMO_MODE', 'false').lower() == 'true'
user = None
# Try multi-user authentication first
# Step 3: Authenticate
if not _is_single_user_mode():
try:
from app.services.user_service import get_user_service
user = get_user_service().authenticate(username, password)
# Check if user has no password set (code-login user)
if user and user.get('_no_password'):
user.pop('_no_password', None)
# Record failed attempt
security.record_login_attempt(ip_address, 'ip', False, ip_address, user_agent)
security.record_login_attempt(username, 'account', False, ip_address, user_agent)
security.log_security_event('login_failed', user.get('id'), ip_address, user_agent,
{'username': username, 'reason': 'no_password_set'})
return jsonify({
'code': 0,
'msg': 'This account was created with email verification code and has no password set. Please use email code login or set a password first in your profile settings.',
'data': None
}), 401
except Exception as e:
logger.warning(f"Multi-user auth failed, trying legacy: {e}")
@@ -60,9 +136,23 @@ def login():
user = authenticate_legacy(username, password)
if not user:
# Record failed attempt
security.record_login_attempt(ip_address, 'ip', False, ip_address, user_agent)
security.record_login_attempt(username, 'account', False, ip_address, user_agent)
security.log_security_event('login_failed', None, ip_address, user_agent,
{'username': username, 'reason': 'invalid_credentials'})
return jsonify({'code': 0, 'msg': 'Invalid credentials', 'data': None}), 401
# Generate token
# Check user status
if user.get('status') == 'disabled':
security.log_security_event('login_blocked', user.get('id'), ip_address, user_agent,
{'reason': 'account_disabled'})
return jsonify({'code': 0, 'msg': 'Account is disabled', 'data': None}), 403
if user.get('status') == 'pending':
return jsonify({'code': 0, 'msg': 'Account is pending activation', 'data': None}), 403
# Step 4: Generate token
token = generate_token(
user_id=user.get('id') or user.get('user_id', 1),
username=user.get('username', username),
@@ -72,6 +162,13 @@ def login():
if not token:
return jsonify({'code': 500, 'msg': 'Token generation error', 'data': None}), 500
# Step 5: Record successful login
security.record_login_attempt(ip_address, 'ip', True, ip_address, user_agent)
security.record_login_attempt(username, 'account', True, ip_address, user_agent)
security.clear_login_attempts(ip_address, 'ip')
security.clear_login_attempts(username, 'account')
security.log_security_event('login_success', user.get('id'), ip_address, user_agent)
# Build user info for frontend
userinfo = {
'id': user.get('id') or user.get('user_id', 1),
@@ -99,6 +196,741 @@ def login():
return jsonify({'code': 500, 'msg': str(e), 'data': None}), 500
# =============================================================================
# Email Code Login
# =============================================================================
@auth_bp.route('/login-code', methods=['POST'])
def login_with_code():
"""
Login with email verification code (quick login / register).
If user doesn't exist, create a new account automatically.
Request body:
email: str
code: str (verification code)
turnstile_token: str (optional)
referral_code: str (optional, referrer's user ID - only for new users)
"""
ip_address = _get_client_ip()
user_agent = _get_user_agent()
try:
from app.services.security_service import get_security_service
from app.services.email_service import get_email_service
from app.services.user_service import get_user_service
from app.services.billing_service import get_billing_service
security = get_security_service()
email_service = get_email_service()
user_service = get_user_service()
billing_service = get_billing_service()
data = request.get_json()
if not data:
return jsonify({'code': 0, 'msg': 'No data provided', 'data': None}), 400
email = (data.get('email') or '').strip().lower()
code = data.get('code', '').strip()
turnstile_token = data.get('turnstile_token')
referral_code = data.get('referral_code', '').strip()
# Validate inputs
if not email or not email_service.is_valid_email(email):
return jsonify({'code': 0, 'msg': 'Invalid email address', 'data': None}), 400
if not code:
return jsonify({'code': 0, 'msg': 'Verification code is required', 'data': None}), 400
# Verify Turnstile
turnstile_ok, turnstile_msg = security.verify_turnstile(turnstile_token, ip_address)
if not turnstile_ok:
return jsonify({'code': 0, 'msg': turnstile_msg, 'data': None}), 400
# Verify email code
code_valid, code_msg = email_service.verify_code(email, code, 'login')
if not code_valid:
return jsonify({'code': 0, 'msg': code_msg, 'data': None}), 400
# Check if user exists
user = user_service.get_user_by_email(email)
is_new_user = False
if not user:
# Check if registration is enabled
if os.getenv('ENABLE_REGISTRATION', 'true').lower() != 'true':
return jsonify({'code': 0, 'msg': 'User not found and registration is disabled', 'data': None}), 403
# Auto-create user with email as username
import re
# Generate username from email (before @)
base_username = re.sub(r'[^a-zA-Z0-9_]', '', email.split('@')[0])
if not base_username or not base_username[0].isalpha():
base_username = 'user_' + base_username
# Make sure username is unique
username = base_username
counter = 1
while user_service.get_user_by_username(username):
username = f"{base_username}_{counter}"
counter += 1
# Validate referral code (user ID)
referred_by = None
if referral_code:
try:
referrer_id = int(referral_code)
referrer = user_service.get_user_by_id(referrer_id)
if referrer and referrer.get('status') == 'active':
referred_by = referrer_id
except (ValueError, TypeError):
pass # Invalid referral code, ignore
# Create user without password (can set later)
user_id = user_service.create_user(
username=username,
password=None, # No password for code-login users
email=email,
nickname=username,
role='user',
status='active',
email_verified=True,
referred_by=referred_by
)
if not user_id:
return jsonify({'code': 0, 'msg': 'Failed to create account', 'data': None}), 500
# Grant registration bonus credits
register_bonus = int(os.getenv('CREDITS_REGISTER_BONUS', '0'))
if register_bonus > 0:
billing_service.add_credits(
user_id=user_id,
amount=register_bonus,
action='register_bonus',
remark='Registration bonus'
)
# Grant referral bonus to referrer
if referred_by:
referral_bonus = int(os.getenv('CREDITS_REFERRAL_BONUS', '0'))
if referral_bonus > 0:
billing_service.add_credits(
user_id=referred_by,
amount=referral_bonus,
action='referral_bonus',
remark=f'Referral bonus for inviting user {username}',
reference_id=str(user_id)
)
user = user_service.get_user_by_id(user_id)
is_new_user = True
# Log registration
security.log_security_event('register_via_code', user_id, ip_address, user_agent,
{'email': email, 'referred_by': referred_by})
# Check user status
if user.get('status') == 'disabled':
security.log_security_event('login_blocked', user.get('id'), ip_address, user_agent,
{'reason': 'account_disabled'})
return jsonify({'code': 0, 'msg': 'Account is disabled', 'data': None}), 403
# Generate token
token = generate_token(
user_id=user['id'],
username=user['username'],
role=user.get('role', 'user')
)
if not token:
return jsonify({'code': 500, 'msg': 'Token generation error', 'data': None}), 500
# Update last login time
try:
from app.utils.db import get_db_connection
with get_db_connection() as db:
cur = db.cursor()
cur.execute(
"UPDATE qd_users SET last_login_at = NOW() WHERE id = ?",
(user['id'],)
)
db.commit()
cur.close()
except Exception as e:
logger.warning(f"Failed to update last_login_at: {e}")
# Log login
security.log_security_event('login_via_code', user['id'], ip_address, user_agent)
is_demo = os.getenv('IS_DEMO_MODE', 'false').lower() == 'true'
return jsonify({
'code': 1,
'msg': 'Login successful' + (' (new account created)' if is_new_user else ''),
'data': {
'token': token,
'is_new_user': is_new_user,
'userinfo': {
'id': user['id'],
'username': user['username'],
'nickname': user.get('nickname', user['username']) + (' (Demo)' if is_demo else ''),
'email': user.get('email'),
'avatar': user.get('avatar', '/avatar2.jpg'),
'is_demo': is_demo,
'role': {
'id': user.get('role', 'user'),
'permissions': _get_permissions(user.get('role', 'user'))
}
}
}
})
except Exception as e:
logger.error(f"login_with_code error: {e}")
return jsonify({'code': 0, 'msg': 'Login failed', 'data': None}), 500
# =============================================================================
# Registration Endpoints
# =============================================================================
@auth_bp.route('/send-code', methods=['POST'])
def send_verification_code():
"""
Send verification code to email.
Request body:
email: str
type: str (register, reset_password, change_password, change_email)
turnstile_token: str (optional)
"""
ip_address = _get_client_ip()
try:
from app.services.security_service import get_security_service
from app.services.email_service import get_email_service
security = get_security_service()
email_service = get_email_service()
data = request.get_json()
if not data:
return jsonify({'code': 0, 'msg': 'No data provided', 'data': None}), 400
email = (data.get('email') or '').strip().lower()
code_type = data.get('type', 'register')
turnstile_token = data.get('turnstile_token')
# Validate email
if not email or not email_service.is_valid_email(email):
return jsonify({'code': 0, 'msg': 'Invalid email address', 'data': None}), 400
# Verify Turnstile
turnstile_ok, turnstile_msg = security.verify_turnstile(turnstile_token, ip_address)
if not turnstile_ok:
return jsonify({'code': 0, 'msg': turnstile_msg, 'data': None}), 400
# Check rate limit
can_send, rate_msg = security.can_send_verification_code(email, ip_address)
if not can_send:
return jsonify({'code': 0, 'msg': rate_msg, 'data': None}), 429
# For registration, check if email already exists
if code_type == 'register':
from app.services.user_service import get_user_service
existing = get_user_service().get_user_by_email(email)
if existing:
return jsonify({'code': 0, 'msg': 'Email already registered', 'data': None}), 400
# For login type - always allow (will auto-create if not exists)
# No special check needed
# For reset_password, check if email exists
if code_type == 'reset_password':
from app.services.user_service import get_user_service
existing = get_user_service().get_user_by_email(email)
if not existing:
# Don't reveal if email exists or not (security best practice)
# But still return success to prevent email enumeration
return jsonify({'code': 1, 'msg': 'If the email exists, a verification code has been sent', 'data': None})
# Send verification code
success, msg = email_service.send_verification_code(email, code_type, ip_address)
if success:
security.log_security_event('verification_code_sent', None, ip_address,
_get_user_agent(), {'email': email, 'type': code_type})
return jsonify({'code': 1, 'msg': 'Verification code sent', 'data': None})
else:
return jsonify({'code': 0, 'msg': msg, 'data': None}), 500
except Exception as e:
logger.error(f"send_verification_code error: {e}")
return jsonify({'code': 0, 'msg': 'Failed to send verification code', 'data': None}), 500
@auth_bp.route('/register', methods=['POST'])
def register():
"""
Register new user with email verification.
Request body:
email: str
code: str (verification code)
username: str
password: str
turnstile_token: str (optional)
referral_code: str (optional, referrer's user ID)
"""
ip_address = _get_client_ip()
user_agent = _get_user_agent()
try:
# Check if registration is enabled
if os.getenv('ENABLE_REGISTRATION', 'true').lower() != 'true':
return jsonify({'code': 0, 'msg': 'Registration is disabled', 'data': None}), 403
from app.services.security_service import get_security_service
from app.services.email_service import get_email_service
from app.services.user_service import get_user_service
from app.services.billing_service import get_billing_service
security = get_security_service()
email_service = get_email_service()
user_service = get_user_service()
billing_service = get_billing_service()
data = request.get_json()
if not data:
return jsonify({'code': 0, 'msg': 'No data provided', 'data': None}), 400
email = (data.get('email') or '').strip().lower()
code = data.get('code', '').strip()
username = (data.get('username') or '').strip()
password = data.get('password', '')
turnstile_token = data.get('turnstile_token')
referral_code = data.get('referral_code', '').strip()
# Validate inputs
if not email or not email_service.is_valid_email(email):
return jsonify({'code': 0, 'msg': 'Invalid email address', 'data': None}), 400
if not code:
return jsonify({'code': 0, 'msg': 'Verification code is required', 'data': None}), 400
if not username or len(username) < 3 or len(username) > 30:
return jsonify({'code': 0, 'msg': 'Username must be 3-30 characters', 'data': None}), 400
# Validate username format (alphanumeric and underscore only)
import re
if not re.match(r'^[a-zA-Z][a-zA-Z0-9_]*$', username):
return jsonify({'code': 0, 'msg': 'Username must start with letter and contain only letters, numbers, and underscores', 'data': None}), 400
# Validate password strength
pwd_valid, pwd_msg = security.validate_password_strength(password)
if not pwd_valid:
return jsonify({'code': 0, 'msg': pwd_msg, 'data': None}), 400
# Verify Turnstile
turnstile_ok, turnstile_msg = security.verify_turnstile(turnstile_token, ip_address)
if not turnstile_ok:
return jsonify({'code': 0, 'msg': turnstile_msg, 'data': None}), 400
# Verify email code
code_valid, code_msg = email_service.verify_code(email, code, 'register')
if not code_valid:
return jsonify({'code': 0, 'msg': code_msg, 'data': None}), 400
# Check if username already exists
existing_user = user_service.get_user_by_username(username)
if existing_user:
return jsonify({'code': 0, 'msg': 'Username already taken', 'data': None}), 400
# Check if email already exists
existing_email = user_service.get_user_by_email(email)
if existing_email:
return jsonify({'code': 0, 'msg': 'Email already registered', 'data': None}), 400
# Validate referral code (user ID)
referred_by = None
if referral_code:
try:
referrer_id = int(referral_code)
referrer = user_service.get_user_by_id(referrer_id)
if referrer and referrer.get('status') == 'active':
referred_by = referrer_id
except (ValueError, TypeError):
pass # Invalid referral code, ignore
# Create user
user_id = user_service.create_user(
username=username,
password=password,
email=email,
nickname=username,
role='user',
status='active',
email_verified=True,
referred_by=referred_by
)
if not user_id:
return jsonify({'code': 0, 'msg': 'Failed to create account', 'data': None}), 500
# Grant registration bonus credits
register_bonus = int(os.getenv('CREDITS_REGISTER_BONUS', '0'))
if register_bonus > 0:
billing_service.add_credits(
user_id=user_id,
amount=register_bonus,
action='register_bonus',
remark='Registration bonus'
)
# Grant referral bonus to referrer
if referred_by:
referral_bonus = int(os.getenv('CREDITS_REFERRAL_BONUS', '0'))
if referral_bonus > 0:
billing_service.add_credits(
user_id=referred_by,
amount=referral_bonus,
action='referral_bonus',
remark=f'Referral bonus for inviting user {username}',
reference_id=str(user_id)
)
# Log registration
security.log_security_event('register', user_id, ip_address, user_agent,
{'email': email, 'referred_by': referred_by})
# Auto login after registration
token = generate_token(user_id=user_id, username=username, role='user')
is_demo = os.getenv('IS_DEMO_MODE', 'false').lower() == 'true'
return jsonify({
'code': 1,
'msg': 'Registration successful',
'data': {
'token': token,
'userinfo': {
'id': user_id,
'username': username,
'nickname': username,
'email': email,
'avatar': '/avatar2.jpg',
'is_demo': is_demo,
'role': {
'id': 'user',
'permissions': _get_permissions('user')
}
}
}
})
except Exception as e:
logger.error(f"register error: {e}")
return jsonify({'code': 0, 'msg': 'Registration failed', 'data': None}), 500
@auth_bp.route('/reset-password', methods=['POST'])
def reset_password():
"""
Reset password with email verification.
Request body:
email: str
code: str (verification code)
new_password: str
turnstile_token: str (optional)
"""
ip_address = _get_client_ip()
user_agent = _get_user_agent()
try:
from app.services.security_service import get_security_service
from app.services.email_service import get_email_service
from app.services.user_service import get_user_service
security = get_security_service()
email_service = get_email_service()
user_service = get_user_service()
data = request.get_json()
if not data:
return jsonify({'code': 0, 'msg': 'No data provided', 'data': None}), 400
email = (data.get('email') or '').strip().lower()
code = data.get('code', '').strip()
new_password = data.get('new_password', '')
turnstile_token = data.get('turnstile_token')
# Validate inputs
if not email or not code or not new_password:
return jsonify({'code': 0, 'msg': 'Missing required fields', 'data': None}), 400
# Validate password strength
pwd_valid, pwd_msg = security.validate_password_strength(new_password)
if not pwd_valid:
return jsonify({'code': 0, 'msg': pwd_msg, 'data': None}), 400
# Verify Turnstile
turnstile_ok, turnstile_msg = security.verify_turnstile(turnstile_token, ip_address)
if not turnstile_ok:
return jsonify({'code': 0, 'msg': turnstile_msg, 'data': None}), 400
# Verify email code
code_valid, code_msg = email_service.verify_code(email, code, 'reset_password')
if not code_valid:
return jsonify({'code': 0, 'msg': code_msg, 'data': None}), 400
# Get user by email
user = user_service.get_user_by_email(email)
if not user:
return jsonify({'code': 0, 'msg': 'User not found', 'data': None}), 404
# Update password
success = user_service.update_password(user['id'], new_password)
if not success:
return jsonify({'code': 0, 'msg': 'Failed to reset password', 'data': None}), 500
# Clear any existing login blocks for this account
security.clear_login_attempts(user['username'], 'account')
# Log password reset
security.log_security_event('password_reset', user['id'], ip_address, user_agent)
return jsonify({'code': 1, 'msg': 'Password reset successful', 'data': None})
except Exception as e:
logger.error(f"reset_password error: {e}")
return jsonify({'code': 0, 'msg': 'Password reset failed', 'data': None}), 500
@auth_bp.route('/change-password', methods=['POST'])
@login_required
def change_password():
"""
Change password with email verification (for logged-in users).
Request body:
code: str (verification code sent to user's email)
new_password: str
"""
ip_address = _get_client_ip()
user_agent = _get_user_agent()
user_id = g.user_id
try:
from app.services.security_service import get_security_service
from app.services.email_service import get_email_service
from app.services.user_service import get_user_service
security = get_security_service()
email_service = get_email_service()
user_service = get_user_service()
data = request.get_json()
if not data:
return jsonify({'code': 0, 'msg': 'No data provided', 'data': None}), 400
code = data.get('code', '').strip()
new_password = data.get('new_password', '')
if not code or not new_password:
return jsonify({'code': 0, 'msg': 'Missing required fields', 'data': None}), 400
# Validate password strength
pwd_valid, pwd_msg = security.validate_password_strength(new_password)
if not pwd_valid:
return jsonify({'code': 0, 'msg': pwd_msg, 'data': None}), 400
# Get user
user = user_service.get_user_by_id(user_id)
if not user or not user.get('email'):
return jsonify({'code': 0, 'msg': 'User email not found', 'data': None}), 400
# Verify email code
code_valid, code_msg = email_service.verify_code(user['email'], code, 'change_password')
if not code_valid:
return jsonify({'code': 0, 'msg': code_msg, 'data': None}), 400
# Update password
success = user_service.update_password(user_id, new_password)
if not success:
return jsonify({'code': 0, 'msg': 'Failed to change password', 'data': None}), 500
# Log password change
security.log_security_event('password_changed', user_id, ip_address, user_agent)
return jsonify({'code': 1, 'msg': 'Password changed successfully', 'data': None})
except Exception as e:
logger.error(f"change_password error: {e}")
return jsonify({'code': 0, 'msg': 'Password change failed', 'data': None}), 500
# =============================================================================
# OAuth Endpoints
# =============================================================================
@auth_bp.route('/oauth/google', methods=['GET'])
def oauth_google():
"""Redirect to Google OAuth authorization page"""
try:
from app.services.oauth_service import get_oauth_service
oauth = get_oauth_service()
if not oauth.google_enabled:
return jsonify({'code': 0, 'msg': 'Google OAuth is not configured', 'data': None}), 400
auth_url, state = oauth.get_google_auth_url()
return redirect(auth_url)
except Exception as e:
logger.error(f"oauth_google error: {e}")
return jsonify({'code': 0, 'msg': str(e), 'data': None}), 500
@auth_bp.route('/oauth/google/callback', methods=['GET'])
def oauth_google_callback():
"""Handle Google OAuth callback"""
ip_address = _get_client_ip()
user_agent = _get_user_agent()
try:
from app.services.oauth_service import get_oauth_service
from app.services.security_service import get_security_service
oauth = get_oauth_service()
security = get_security_service()
code = request.args.get('code')
state = request.args.get('state')
error = request.args.get('error')
frontend_url = oauth.frontend_url
if error:
return redirect(f"{frontend_url}/user/login?oauth_error={error}")
if not code or not state:
return redirect(f"{frontend_url}/user/login?oauth_error=missing_params")
# Handle callback
success, result = oauth.handle_google_callback(code, state)
if not success:
error_msg = result.get('error', 'unknown_error')
return redirect(f"{frontend_url}/user/login?oauth_error={error_msg}")
# Get or create user
user_success, user_result = oauth.get_or_create_user_from_oauth(result)
if not user_success:
error_msg = user_result.get('error', 'user_creation_failed')
return redirect(f"{frontend_url}/user/login?oauth_error={error_msg}")
# Generate token
token = generate_token(
user_id=user_result['id'],
username=user_result['username'],
role=user_result.get('role', 'user')
)
# Log OAuth login
security.log_security_event('oauth_login', user_result['id'], ip_address, user_agent,
{'provider': 'google'})
# Redirect to frontend with token
return redirect(f"{frontend_url}/user/login?oauth_token={token}")
except Exception as e:
logger.error(f"oauth_google_callback error: {e}")
from app.services.oauth_service import get_oauth_service
frontend_url = get_oauth_service().frontend_url
return redirect(f"{frontend_url}/user/login?oauth_error=server_error")
@auth_bp.route('/oauth/github', methods=['GET'])
def oauth_github():
"""Redirect to GitHub OAuth authorization page"""
try:
from app.services.oauth_service import get_oauth_service
oauth = get_oauth_service()
if not oauth.github_enabled:
return jsonify({'code': 0, 'msg': 'GitHub OAuth is not configured', 'data': None}), 400
auth_url, state = oauth.get_github_auth_url()
return redirect(auth_url)
except Exception as e:
logger.error(f"oauth_github error: {e}")
return jsonify({'code': 0, 'msg': str(e), 'data': None}), 500
@auth_bp.route('/oauth/github/callback', methods=['GET'])
def oauth_github_callback():
"""Handle GitHub OAuth callback"""
ip_address = _get_client_ip()
user_agent = _get_user_agent()
try:
from app.services.oauth_service import get_oauth_service
from app.services.security_service import get_security_service
oauth = get_oauth_service()
security = get_security_service()
code = request.args.get('code')
state = request.args.get('state')
error = request.args.get('error')
frontend_url = oauth.frontend_url
if error:
return redirect(f"{frontend_url}/user/login?oauth_error={error}")
if not code or not state:
return redirect(f"{frontend_url}/user/login?oauth_error=missing_params")
# Handle callback
success, result = oauth.handle_github_callback(code, state)
if not success:
error_msg = result.get('error', 'unknown_error')
return redirect(f"{frontend_url}/user/login?oauth_error={error_msg}")
# Get or create user
user_success, user_result = oauth.get_or_create_user_from_oauth(result)
if not user_success:
error_msg = user_result.get('error', 'user_creation_failed')
return redirect(f"{frontend_url}/user/login?oauth_error={error_msg}")
# Generate token
token = generate_token(
user_id=user_result['id'],
username=user_result['username'],
role=user_result.get('role', 'user')
)
# Log OAuth login
security.log_security_event('oauth_login', user_result['id'], ip_address, user_agent,
{'provider': 'github'})
# Redirect to frontend with token
return redirect(f"{frontend_url}/user/login?oauth_token={token}")
except Exception as e:
logger.error(f"oauth_github_callback error: {e}")
from app.services.oauth_service import get_oauth_service
frontend_url = get_oauth_service().frontend_url
return redirect(f"{frontend_url}/user/login?oauth_error=server_error")
# =============================================================================
# Other Endpoints
# =============================================================================
@auth_bp.route('/logout', methods=['POST'])
def logout():
"""Logout (client removes token; server is stateless)."""
backend_api_python/app/routes/portfolio.py
+42 -7
View File
@@ -680,25 +680,60 @@ def delete_monitor(monitor_id):
@portfolio_bp.route('/monitors/<int:monitor_id>/run', methods=['POST'])
@login_required
def run_monitor_now(monitor_id):
"""Manually trigger a monitor to run immediately."""
"""Manually trigger a monitor to run immediately.
Supports two modes:
- async=true (default): Returns immediately, runs in background, notifies via notification system
- async=false: Waits for completion and returns result (may timeout for large portfolios)
"""
try:
from app.services.portfolio_monitor import run_single_monitor
# Get language from request body or Accept-Language header
user_id = g.user_id
# Get parameters from request body
data = request.get_json(force=True, silent=True) or {}
language = data.get('language')
async_mode = data.get('async', True) # Default to async mode
# Fallback to Accept-Language header
# Fallback to Accept-Language header for language
if not language:
accept_lang = request.headers.get('Accept-Language', '')
if 'zh' in accept_lang.lower():
language = 'en-US'
language = 'zh-CN'
else:
language = 'en-US'
result = run_single_monitor(monitor_id, override_language=language)
return jsonify({'code': 1, 'msg': 'success', 'data': result})
if async_mode:
# Async mode: Start background thread and return immediately
import threading
def run_in_background(mid, lang, uid):
try:
run_single_monitor(mid, override_language=lang, user_id=uid)
except Exception as e:
logger.error(f"Background monitor run failed: {e}")
thread = threading.Thread(
target=run_in_background,
args=(monitor_id, language, user_id),
daemon=True
)
thread.start()
return jsonify({
'code': 1,
'msg': 'success',
'data': {
'status': 'running',
'message': 'Monitor is running in background. Results will be sent via notification.'
}
})
else:
# Sync mode: Wait for completion (may timeout)
result = run_single_monitor(monitor_id, override_language=language, user_id=user_id)
return jsonify({'code': 1, 'msg': 'success', 'data': result})
except Exception as e:
logger.error(f"run_monitor_now failed: {str(e)}")
logger.error(traceback.format_exc())
backend_api_python/app/routes/settings.py
+243 -2
View File
@@ -76,6 +76,13 @@ CONFIG_SCHEMA = {
'default': '123456',
'description': 'Administrator login password. MUST change in production'
},
{
'key': 'ADMIN_EMAIL',
'label': 'Admin Email',
'type': 'text',
'default': 'admin@example.com',
'description': 'Administrator email for password reset and notifications'
},
]
},
@@ -657,11 +664,245 @@ CONFIG_SCHEMA = {
]
},
# ==================== 13. 应用配置 ====================
# ==================== 13. 注册与安全 ====================
'security': {
'title': 'Registration & Security',
'icon': 'safety',
'order': 13,
'items': [
{
'key': 'ENABLE_REGISTRATION',
'label': 'Enable Registration',
'type': 'boolean',
'default': 'True',
'description': 'Allow new users to register accounts'
},
{
'key': 'TURNSTILE_SITE_KEY',
'label': 'Turnstile Site Key',
'type': 'text',
'required': False,
'link': 'https://dash.cloudflare.com/?to=/:account/turnstile',
'link_text': 'settings.link.getTurnstileKey',
'description': 'Cloudflare Turnstile site key for CAPTCHA verification'
},
{
'key': 'TURNSTILE_SECRET_KEY',
'label': 'Turnstile Secret Key',
'type': 'password',
'required': False,
'description': 'Cloudflare Turnstile secret key'
},
{
'key': 'FRONTEND_URL',
'label': 'Frontend URL',
'type': 'text',
'default': 'http://localhost:8080',
'description': 'Frontend URL for OAuth redirects'
},
{
'key': 'GOOGLE_CLIENT_ID',
'label': 'Google Client ID',
'type': 'text',
'required': False,
'link': 'https://console.cloud.google.com/apis/credentials',
'link_text': 'settings.link.getGoogleCredentials',
'description': 'Google OAuth Client ID'
},
{
'key': 'GOOGLE_CLIENT_SECRET',
'label': 'Google Client Secret',
'type': 'password',
'required': False,
'description': 'Google OAuth Client Secret'
},
{
'key': 'GOOGLE_REDIRECT_URI',
'label': 'Google Redirect URI',
'type': 'text',
'default': 'http://localhost:5000/api/auth/oauth/google/callback',
'description': 'Google OAuth callback URL'
},
{
'key': 'GITHUB_CLIENT_ID',
'label': 'GitHub Client ID',
'type': 'text',
'required': False,
'link': 'https://github.com/settings/developers',
'link_text': 'settings.link.getGithubCredentials',
'description': 'GitHub OAuth Client ID'
},
{
'key': 'GITHUB_CLIENT_SECRET',
'label': 'GitHub Client Secret',
'type': 'password',
'required': False,
'description': 'GitHub OAuth Client Secret'
},
{
'key': 'GITHUB_REDIRECT_URI',
'label': 'GitHub Redirect URI',
'type': 'text',
'default': 'http://localhost:5000/api/auth/oauth/github/callback',
'description': 'GitHub OAuth callback URL'
},
{
'key': 'SECURITY_IP_MAX_ATTEMPTS',
'label': 'IP Max Failed Attempts',
'type': 'number',
'default': '10',
'description': 'Block IP after this many failed login attempts'
},
{
'key': 'SECURITY_IP_WINDOW_MINUTES',
'label': 'IP Window (minutes)',
'type': 'number',
'default': '5',
'description': 'Time window for counting IP failed attempts'
},
{
'key': 'SECURITY_IP_BLOCK_MINUTES',
'label': 'IP Block Duration (minutes)',
'type': 'number',
'default': '15',
'description': 'How long to block IP after exceeding limit'
},
{
'key': 'SECURITY_ACCOUNT_MAX_ATTEMPTS',
'label': 'Account Max Failed Attempts',
'type': 'number',
'default': '5',
'description': 'Lock account after this many failed login attempts'
},
{
'key': 'SECURITY_ACCOUNT_WINDOW_MINUTES',
'label': 'Account Window (minutes)',
'type': 'number',
'default': '60',
'description': 'Time window for counting account failed attempts'
},
{
'key': 'SECURITY_ACCOUNT_BLOCK_MINUTES',
'label': 'Account Block Duration (minutes)',
'type': 'number',
'default': '30',
'description': 'How long to lock account after exceeding limit'
},
{
'key': 'VERIFICATION_CODE_EXPIRE_MINUTES',
'label': 'Verification Code Expiry (minutes)',
'type': 'number',
'default': '10',
'description': 'Email verification code validity period'
},
{
'key': 'VERIFICATION_CODE_RATE_LIMIT',
'label': 'Code Rate Limit (seconds)',
'type': 'number',
'default': '60',
'description': 'Minimum time between verification code requests per email'
},
{
'key': 'VERIFICATION_CODE_IP_HOURLY_LIMIT',
'label': 'Code Hourly Limit per IP',
'type': 'number',
'default': '10',
'description': 'Maximum verification codes per IP per hour'
},
{
'key': 'VERIFICATION_CODE_MAX_ATTEMPTS',
'label': 'Code Max Attempts',
'type': 'number',
'default': '5',
'description': 'Maximum attempts to verify a code before lockout'
},
{
'key': 'VERIFICATION_CODE_LOCK_MINUTES',
'label': 'Code Lock Minutes',
'type': 'number',
'default': '30',
'description': 'Lockout duration after exceeding max attempts'
},
]
},
# ==================== 14. 计费配置 ====================
'billing': {
'title': 'Billing & Credits',
'icon': 'dollar',
'order': 14,
'items': [
{
'key': 'BILLING_ENABLED',
'label': 'Enable Billing',
'type': 'boolean',
'default': 'False',
'description': 'Enable billing system. When enabled, users need credits to use certain features'
},
{
'key': 'BILLING_VIP_BYPASS',
'label': 'VIP Free',
'type': 'boolean',
'default': 'True',
'description': 'VIP users can use all paid features for free during VIP period'
},
{
'key': 'BILLING_COST_AI_ANALYSIS',
'label': 'AI Analysis Cost',
'type': 'number',
'default': '10',
'description': 'Credits consumed per AI analysis request'
},
{
'key': 'BILLING_COST_STRATEGY_RUN',
'label': 'Strategy Run Cost',
'type': 'number',
'default': '5',
'description': 'Credits consumed when starting a strategy'
},
{
'key': 'BILLING_COST_BACKTEST',
'label': 'Backtest Cost',
'type': 'number',
'default': '3',
'description': 'Credits consumed per backtest run'
},
{
'key': 'BILLING_COST_PORTFOLIO_MONITOR',
'label': 'Portfolio Monitor Cost',
'type': 'number',
'default': '8',
'description': 'Credits consumed per portfolio AI monitoring run'
},
{
'key': 'RECHARGE_TELEGRAM_URL',
'label': 'Recharge Telegram URL',
'type': 'text',
'default': 'https://t.me/your_support_bot',
'description': 'Telegram customer service URL for recharge inquiries'
},
{
'key': 'CREDITS_REGISTER_BONUS',
'label': 'Register Bonus',
'type': 'number',
'default': '100',
'description': 'Credits awarded to new users on registration'
},
{
'key': 'CREDITS_REFERRAL_BONUS',
'label': 'Referral Bonus',
'type': 'number',
'default': '50',
'description': 'Credits awarded to referrer when someone signs up with their code'
},
]
},
# ==================== 15. 应用配置 ====================
'app': {
'title': 'Application',
'icon': 'appstore',
'order': 13,
'order': 15,
'items': [
{
'key': 'CORS_ORIGINS',
backend_api_python/app/routes/user.py
+298 -10
View File
@@ -24,13 +24,15 @@ def list_users():
Query params:
page: int (default 1)
page_size: int (default 20, max 100)
search: str (optional, search by username/email/nickname)
"""
try:
page = request.args.get('page', 1, type=int)
page_size = request.args.get('page_size', 20, type=int)
search = request.args.get('search', '', type=str)
page_size = min(100, max(1, page_size))
result = get_user_service().list_users(page=page, page_size=page_size)
result = get_user_service().list_users(page=page, page_size=page_size, search=search)
return jsonify({
'code': 1,
@@ -213,13 +215,143 @@ def get_roles():
})
# ==================== Billing Management (Admin) ====================
@user_bp.route('/set-credits', methods=['POST'])
@login_required
@admin_required
def set_user_credits():
"""
Set user credits (admin only).
Request body:
user_id: int (required)
credits: int (required)
remark: str (optional)
"""
try:
from app.services.billing_service import get_billing_service
data = request.get_json() or {}
user_id = data.get('user_id')
credits = data.get('credits')
remark = data.get('remark', '')
if not user_id:
return jsonify({'code': 0, 'msg': 'Missing user_id', 'data': None}), 400
if credits is None or credits < 0:
return jsonify({'code': 0, 'msg': 'Credits must be a non-negative number', 'data': None}), 400
operator_id = getattr(g, 'user_id', None)
success, result = get_billing_service().set_credits(user_id, int(credits), remark, operator_id)
if success:
return jsonify({'code': 1, 'msg': 'Credits updated successfully', 'data': {'credits': result}})
else:
return jsonify({'code': 0, 'msg': result, 'data': None}), 400
except Exception as e:
logger.error(f"set_user_credits failed: {e}")
return jsonify({'code': 0, 'msg': str(e), 'data': None}), 500
@user_bp.route('/set-vip', methods=['POST'])
@login_required
@admin_required
def set_user_vip():
"""
Set user VIP status (admin only).
Request body:
user_id: int (required)
vip_days: int (optional, 0 to cancel VIP, positive number to grant VIP for days)
vip_expires_at: str (optional, ISO format datetime, overrides vip_days if provided)
remark: str (optional)
"""
try:
from datetime import datetime, timedelta, timezone
from app.services.billing_service import get_billing_service
data = request.get_json() or {}
user_id = data.get('user_id')
vip_days = data.get('vip_days')
vip_expires_at_str = data.get('vip_expires_at')
remark = data.get('remark', '')
if not user_id:
return jsonify({'code': 0, 'msg': 'Missing user_id', 'data': None}), 400
# Calculate expires_at
expires_at = None
if vip_expires_at_str:
try:
expires_at = datetime.fromisoformat(vip_expires_at_str.replace('Z', '+00:00'))
except ValueError:
return jsonify({'code': 0, 'msg': 'Invalid vip_expires_at format', 'data': None}), 400
elif vip_days is not None:
if vip_days > 0:
expires_at = datetime.now(timezone.utc) + timedelta(days=vip_days)
else:
expires_at = None # Cancel VIP
else:
return jsonify({'code': 0, 'msg': 'Provide vip_days or vip_expires_at', 'data': None}), 400
operator_id = getattr(g, 'user_id', None)
success, result = get_billing_service().set_vip(user_id, expires_at, remark, operator_id)
if success:
return jsonify({
'code': 1,
'msg': 'VIP status updated successfully',
'data': {'vip_expires_at': expires_at.isoformat() if expires_at else None}
})
else:
return jsonify({'code': 0, 'msg': result, 'data': None}), 400
except Exception as e:
logger.error(f"set_user_vip failed: {e}")
return jsonify({'code': 0, 'msg': str(e), 'data': None}), 500
@user_bp.route('/credits-log', methods=['GET'])
@login_required
@admin_required
def get_user_credits_log():
"""
Get user credits log (admin only).
Query params:
user_id: int (required)
page: int (default 1)
page_size: int (default 20)
"""
try:
from app.services.billing_service import get_billing_service
user_id = request.args.get('user_id', type=int)
if not user_id:
return jsonify({'code': 0, 'msg': 'Missing user_id', 'data': None}), 400
page = request.args.get('page', 1, type=int)
page_size = request.args.get('page_size', 20, type=int)
page_size = min(100, max(1, page_size))
result = get_billing_service().get_credits_log(user_id, page, page_size)
return jsonify({'code': 1, 'msg': 'success', 'data': result})
except Exception as e:
logger.error(f"get_user_credits_log failed: {e}")
return jsonify({'code': 0, 'msg': str(e), 'data': None}), 500
# Self-service endpoints (accessible by any logged-in user)
@user_bp.route('/profile', methods=['GET'])
@login_required
def get_profile():
"""Get current user's profile"""
"""Get current user's profile with billing info"""
try:
from app.services.billing_service import get_billing_service
user_id = getattr(g, 'user_id', None)
if not user_id:
return jsonify({'code': 0, 'msg': 'Not authenticated', 'data': None}), 401
@@ -231,6 +363,10 @@ def get_profile():
# Add permissions
user['permissions'] = get_user_service().get_user_permissions(user.get('role', 'user'))
# Add billing info
billing_info = get_billing_service().get_user_billing_info(user_id)
user['billing'] = billing_info
return jsonify({
'code': 1,
'msg': 'success',
@@ -249,8 +385,10 @@ def update_profile():
Request body:
nickname: str (optional)
email: str (optional)
avatar: str (optional)
Note: Email cannot be changed after registration (for security).
Only admin can change user email via User Management.
"""
try:
user_id = getattr(g, 'user_id', None)
@@ -260,8 +398,9 @@ def update_profile():
data = request.get_json() or {}
# Only allow updating certain fields for self-service
# Email is NOT allowed to be changed (security: bound to account)
allowed = {}
for field in ['nickname', 'email', 'avatar']:
for field in ['nickname', 'avatar']:
if field in data:
allowed[field] = data[field]
@@ -279,6 +418,117 @@ def update_profile():
return jsonify({'code': 0, 'msg': str(e), 'data': None}), 500
@user_bp.route('/my-credits-log', methods=['GET'])
@login_required
def get_my_credits_log():
"""
Get current user's credits log.
Query params:
page: int (default 1)
page_size: int (default 20)
"""
try:
from app.services.billing_service import get_billing_service
user_id = getattr(g, 'user_id', None)
if not user_id:
return jsonify({'code': 0, 'msg': 'Not authenticated', 'data': None}), 401
page = request.args.get('page', 1, type=int)
page_size = request.args.get('page_size', 20, type=int)
page_size = min(100, max(1, page_size))
result = get_billing_service().get_credits_log(user_id, page, page_size)
return jsonify({'code': 1, 'msg': 'success', 'data': result})
except Exception as e:
logger.error(f"get_my_credits_log failed: {e}")
return jsonify({'code': 0, 'msg': str(e), 'data': None}), 500
@user_bp.route('/my-referrals', methods=['GET'])
@login_required
def get_my_referrals():
"""
Get list of users referred by current user.
Query params:
page: int (default 1)
page_size: int (default 20)
Returns:
list: Users referred by current user (id, username, nickname, avatar, created_at)
total: Total count of referrals
referral_code: Current user's referral code (user ID)
referral_bonus: Credits earned per referral
register_bonus: Credits new users get on registration
"""
try:
import os
from app.utils.db import get_db_connection
user_id = getattr(g, 'user_id', None)
if not user_id:
return jsonify({'code': 0, 'msg': 'Not authenticated', 'data': None}), 401
page = request.args.get('page', 1, type=int)
page_size = request.args.get('page_size', 20, type=int)
page_size = min(100, max(1, page_size))
offset = (page - 1) * page_size
with get_db_connection() as db:
cur = db.cursor()
# Get total count
cur.execute(
"SELECT COUNT(*) as cnt FROM qd_users WHERE referred_by = ?",
(user_id,)
)
total = cur.fetchone()['cnt']
# Get referral list
cur.execute(
"""
SELECT id, username, nickname, avatar, created_at
FROM qd_users
WHERE referred_by = ?
ORDER BY created_at DESC
LIMIT ? OFFSET ?
""",
(user_id, page_size, offset)
)
rows = cur.fetchall()
cur.close()
referrals = []
for row in rows:
referrals.append({
'id': row['id'],
'username': row['username'],
'nickname': row['nickname'],
'avatar': row['avatar'],
'created_at': row['created_at'].isoformat() if row['created_at'] else None
})
return jsonify({
'code': 1,
'msg': 'success',
'data': {
'list': referrals,
'total': total,
'page': page,
'page_size': page_size,
'referral_code': str(user_id),
'referral_bonus': int(os.getenv('CREDITS_REFERRAL_BONUS', '0')),
'register_bonus': int(os.getenv('CREDITS_REGISTER_BONUS', '0'))
}
})
except Exception as e:
logger.error(f"get_my_referrals failed: {e}")
return jsonify({'code': 0, 'msg': str(e), 'data': None}), 500
@user_bp.route('/change-password', methods=['POST'])
@login_required
def change_password():
@@ -298,18 +548,56 @@ def change_password():
old_password = data.get('old_password', '')
new_password = data.get('new_password', '')
if not old_password or not new_password:
return jsonify({'code': 0, 'msg': 'Both old and new password required', 'data': None}), 400
if not new_password:
return jsonify({'code': 0, 'msg': 'New password required', 'data': None}), 400
if len(new_password) < 6:
return jsonify({'code': 0, 'msg': 'New password must be at least 6 characters', 'data': None}), 400
success = get_user_service().change_password(user_id, old_password, new_password)
# Check if user has a password set
user_service = get_user_service()
user = user_service.get_user_by_id(user_id)
if not user:
return jsonify({'code': 0, 'msg': 'User not found', 'data': None}), 404
if success:
return jsonify({'code': 1, 'msg': 'Password changed successfully', 'data': None})
# Get password_hash to check if user has no password
from app.utils.db import get_db_connection
with get_db_connection() as db:
cur = db.cursor()
cur.execute("SELECT password_hash FROM qd_users WHERE id = ?", (user_id,))
row = cur.fetchone()
cur.close()
password_hash = row.get('password_hash', '') if row else ''
has_password = password_hash and password_hash.strip() != ''
# If user has no password, allow setting password without old password
if not has_password:
if not old_password:
# No old password required for users without password
success = user_service.reset_password(user_id, new_password)
if success:
return jsonify({'code': 1, 'msg': 'Password set successfully', 'data': None})
else:
return jsonify({'code': 0, 'msg': 'Failed to set password', 'data': None}), 500
else:
# If old_password is provided but user has no password, ignore it
success = user_service.reset_password(user_id, new_password)
if success:
return jsonify({'code': 1, 'msg': 'Password set successfully', 'data': None})
else:
return jsonify({'code': 0, 'msg': 'Failed to set password', 'data': None}), 500
else:
return jsonify({'code': 0, 'msg': 'Old password incorrect', 'data': None}), 400
# User has existing password, require old password verification
if not old_password:
return jsonify({'code': 0, 'msg': 'Old password required', 'data': None}), 400
success = user_service.change_password(user_id, old_password, new_password)
if success:
return jsonify({'code': 1, 'msg': 'Password changed successfully', 'data': None})
else:
return jsonify({'code': 0, 'msg': 'Old password incorrect', 'data': None}), 400
except ValueError as e:
return jsonify({'code': 0, 'msg': str(e), 'data': None}), 400
except Exception as e:
backend_api_python/app/services/billing_service.py
+443
View File
@@ -0,0 +1,443 @@
"""
Billing Service - 统一计费服务
管理用户积分消费、VIP状态检查、计费配置等功能。
支持两种计费模式：
1. 积分消耗模式：每次使用功能扣除相应积分
2. VIP免费模式：VIP用户在有效期内免费使用
计费配置存储在 .env 文件中，可通过系统设置界面配置。
"""
import os
import time
from datetime import datetime, timezone
from decimal import Decimal
from typing import Dict, Any, Optional, Tuple
from app.utils.db import get_db_connection
from app.utils.logger import get_logger
logger = get_logger(__name__)
# 功能计费配置键名
BILLING_CONFIG_PREFIX = 'BILLING_'
# 默认计费配置
DEFAULT_BILLING_CONFIG = {
# 全局开关
'enabled': False, # 是否启用计费
'vip_bypass': True, # VIP用户是否免费
# 各功能积分消耗（0表示免费）
'cost_ai_analysis': 10, # AI分析 每次消耗积分
'cost_strategy_run': 5, # 策略运行 每次消耗积分（启动时）
'cost_backtest': 3, # 回测 每次消耗积分
'cost_portfolio_monitor': 8, # Portfolio AI监控 每次消耗积分
'cost_indicator_create': 0, # 创建指标 免费
}
# Feature name mapping (for log recording)
FEATURE_NAMES = {
'ai_analysis': 'AI Analysis',
'strategy_run': 'Strategy Run',
'backtest': 'Backtest',
'portfolio_monitor': 'Portfolio Monitor',
'indicator_create': 'Indicator Create',
}
class BillingService:
"""计费服务类"""
def __init__(self):
self._config_cache = None
self._config_cache_time = 0
self._cache_ttl = 60 # 配置缓存60秒
def get_billing_config(self) -> Dict[str, Any]:
"""获取计费配置"""
now = time.time()
if self._config_cache and (now - self._config_cache_time) < self._cache_ttl:
return self._config_cache
config = {}
for key, default_value in DEFAULT_BILLING_CONFIG.items():
env_key = f'{BILLING_CONFIG_PREFIX}{key.upper()}'
value = os.getenv(env_key)
if value is None or value == '':
config[key] = default_value
elif isinstance(default_value, bool):
config[key] = str(value).lower() in ('true', '1', 'yes')
elif isinstance(default_value, int):
try:
config[key] = int(value)
except (ValueError, TypeError):
config[key] = default_value
else:
config[key] = value
self._config_cache = config
self._config_cache_time = now
return config
def clear_config_cache(self):
"""清除配置缓存"""
self._config_cache = None
self._config_cache_time = 0
def is_billing_enabled(self) -> bool:
"""检查是否启用计费"""
config = self.get_billing_config()
return config.get('enabled', False)
def get_feature_cost(self, feature: str) -> int:
"""获取功能消耗积分"""
config = self.get_billing_config()
cost_key = f'cost_{feature}'
return config.get(cost_key, 0)
def get_user_credits(self, user_id: int) -> Decimal:
"""获取用户积分余额"""
try:
with get_db_connection() as db:
cur = db.cursor()
cur.execute(
"SELECT credits FROM qd_users WHERE id = ?",
(user_id,)
)
row = cur.fetchone()
cur.close()
if row:
return Decimal(str(row.get('credits', 0) or 0))
return Decimal('0')
except Exception as e:
logger.error(f"get_user_credits failed: {e}")
return Decimal('0')
def get_user_vip_status(self, user_id: int) -> Tuple[bool, Optional[datetime]]:
"""
获取用户VIP状态
Returns:
(is_vip, expires_at): VIP是否有效, VIP过期时间
"""
try:
with get_db_connection() as db:
cur = db.cursor()
cur.execute(
"SELECT vip_expires_at FROM qd_users WHERE id = ?",
(user_id,)
)
row = cur.fetchone()
cur.close()
if row and row.get('vip_expires_at'):
expires_at = row['vip_expires_at']
# 确保是 datetime 对象
if isinstance(expires_at, str):
expires_at = datetime.fromisoformat(expires_at.replace('Z', '+00:00'))
# 检查是否过期
now = datetime.now(timezone.utc)
if expires_at.tzinfo is None:
expires_at = expires_at.replace(tzinfo=timezone.utc)
is_vip = expires_at > now
return is_vip, expires_at
return False, None
except Exception as e:
logger.error(f"get_user_vip_status failed: {e}")
return False, None
def check_and_consume(self, user_id: int, feature: str, reference_id: str = '') -> Tuple[bool, str]:
"""
检查并消耗积分
Args:
user_id: 用户ID
feature: 功能名称（ai_analysis/strategy_run/backtest/portfolio_monitor等）
reference_id: 关联ID（可选）
Returns:
(success, message): 是否成功, 提示消息
"""
# 检查是否启用计费
if not self.is_billing_enabled():
return True, 'billing_disabled'
config = self.get_billing_config()
cost = self.get_feature_cost(feature)
# 免费功能
if cost <= 0:
return True, 'free_feature'
# 检查VIP状态
if config.get('vip_bypass', True):
is_vip, _ = self.get_user_vip_status(user_id)
if is_vip:
return True, 'vip_free'
# 检查积分余额
credits = self.get_user_credits(user_id)
if credits < cost:
return False, f'insufficient_credits:{credits}:{cost}'
# 扣除积分
try:
new_balance = credits - Decimal(str(cost))
with get_db_connection() as db:
cur = db.cursor()
# 更新用户积分
cur.execute(
"UPDATE qd_users SET credits = ?, updated_at = NOW() WHERE id = ?",
(float(new_balance), user_id)
)
# 记录日志
feature_name = FEATURE_NAMES.get(feature, feature)
cur.execute(
"""
INSERT INTO qd_credits_log
(user_id, action, amount, balance_after, feature, reference_id, remark, created_at)
VALUES (?, 'consume', ?, ?, ?, ?, ?, NOW())
""",
(user_id, -cost, float(new_balance), feature, reference_id, f'Consume: {feature_name}')
)
db.commit()
cur.close()
logger.info(f"User {user_id} consumed {cost} credits for {feature}, balance: {new_balance}")
return True, 'consumed'
except Exception as e:
logger.error(f"check_and_consume failed: {e}")
return False, f'error:{str(e)}'
def add_credits(self, user_id: int, amount: int, action: str = 'recharge',
remark: str = '', operator_id: int = None, reference_id: str = '') -> Tuple[bool, str]:
"""
增加用户积分
Args:
user_id: 用户ID
amount: 增加金额（正数）
action: 操作类型（recharge/admin_adjust/refund/referral_bonus/register_bonus）
remark: 备注
operator_id: 操作人ID（管理员操作时）
reference_id: 关联ID（如被邀请用户ID、订单号等）
Returns:
(success, message)
"""
if amount <= 0:
return False, 'amount_must_be_positive'
try:
credits = self.get_user_credits(user_id)
new_balance = credits + Decimal(str(amount))
with get_db_connection() as db:
cur = db.cursor()
# 更新用户积分
cur.execute(
"UPDATE qd_users SET credits = ?, updated_at = NOW() WHERE id = ?",
(float(new_balance), user_id)
)
# 记录日志（包含 reference_id）
cur.execute(
"""
INSERT INTO qd_credits_log
(user_id, action, amount, balance_after, remark, operator_id, reference_id, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, NOW())
""",
(user_id, action, amount, float(new_balance), remark, operator_id, reference_id)
)
db.commit()
cur.close()
logger.info(f"User {user_id} added {amount} credits ({action}), balance: {new_balance}")
return True, str(new_balance)
except Exception as e:
logger.error(f"add_credits failed: {e}")
return False, str(e)
def set_credits(self, user_id: int, amount: int, remark: str = '',
operator_id: int = None) -> Tuple[bool, str]:
"""
设置用户积分（管理员直接设置）
Args:
user_id: 用户ID
amount: 设置的金额
remark: 备注
operator_id: 操作人ID
Returns:
(success, message)
"""
if amount < 0:
return False, 'amount_cannot_be_negative'
try:
old_credits = self.get_user_credits(user_id)
diff = Decimal(str(amount)) - old_credits
with get_db_connection() as db:
cur = db.cursor()
# 更新用户积分
cur.execute(
"UPDATE qd_users SET credits = ?, updated_at = NOW() WHERE id = ?",
(amount, user_id)
)
# 记录日志
cur.execute(
"""
INSERT INTO qd_credits_log
(user_id, action, amount, balance_after, remark, operator_id, created_at)
VALUES (?, 'admin_adjust', ?, ?, ?, ?, NOW())
""",
(user_id, float(diff), amount, remark or f'Admin adjust: {old_credits} -> {amount}', operator_id)
)
db.commit()
cur.close()
logger.info(f"User {user_id} credits set to {amount} by admin {operator_id}")
return True, str(amount)
except Exception as e:
logger.error(f"set_credits failed: {e}")
return False, str(e)
def set_vip(self, user_id: int, expires_at: Optional[datetime],
remark: str = '', operator_id: int = None) -> Tuple[bool, str]:
"""
设置用户VIP状态
Args:
user_id: 用户ID
expires_at: VIP过期时间（None表示取消VIP）
remark: 备注
operator_id: 操作人ID
Returns:
(success, message)
"""
try:
with get_db_connection() as db:
cur = db.cursor()
# 更新VIP过期时间
cur.execute(
"UPDATE qd_users SET vip_expires_at = ?, updated_at = NOW() WHERE id = ?",
(expires_at, user_id)
)
# 记录日志
action = 'vip_grant' if expires_at else 'vip_revoke'
log_remark = remark or (f'VIP granted until {expires_at}' if expires_at else 'VIP revoked')
cur.execute(
"""
INSERT INTO qd_credits_log
(user_id, action, amount, balance_after, remark, operator_id, created_at)
VALUES (?, ?, 0, (SELECT credits FROM qd_users WHERE id = ?), ?, ?, NOW())
""",
(user_id, action, user_id, log_remark, operator_id)
)
db.commit()
cur.close()
logger.info(f"User {user_id} VIP set to {expires_at} by admin {operator_id}")
return True, 'success'
except Exception as e:
logger.error(f"set_vip failed: {e}")
return False, str(e)
def get_credits_log(self, user_id: int, page: int = 1, page_size: int = 20) -> Dict[str, Any]:
"""获取用户积分变动日志"""
offset = (page - 1) * page_size
try:
with get_db_connection() as db:
cur = db.cursor()
# 获取总数
cur.execute(
"SELECT COUNT(*) as count FROM qd_credits_log WHERE user_id = ?",
(user_id,)
)
total = cur.fetchone()['count']
# 获取日志
cur.execute(
"""
SELECT id, action, amount, balance_after, feature, reference_id, remark, created_at
FROM qd_credits_log
WHERE user_id = ?
ORDER BY created_at DESC
LIMIT ? OFFSET ?
""",
(user_id, page_size, offset)
)
logs = cur.fetchall() or []
cur.close()
return {
'items': logs,
'total': total,
'page': page,
'page_size': page_size,
'total_pages': (total + page_size - 1) // page_size
}
except Exception as e:
logger.error(f"get_credits_log failed: {e}")
return {'items': [], 'total': 0, 'page': 1, 'page_size': page_size, 'total_pages': 0}
def get_user_billing_info(self, user_id: int) -> Dict[str, Any]:
"""获取用户计费信息（供前端显示）"""
credits = self.get_user_credits(user_id)
is_vip, vip_expires_at = self.get_user_vip_status(user_id)
config = self.get_billing_config()
return {
'credits': float(credits),
'is_vip': is_vip,
'vip_expires_at': vip_expires_at.isoformat() if vip_expires_at else None,
'billing_enabled': config.get('enabled', False),
'vip_bypass': config.get('vip_bypass', True),
# 功能费用（供前端显示）
'feature_costs': {
'ai_analysis': config.get('cost_ai_analysis', 0),
'strategy_run': config.get('cost_strategy_run', 0),
'backtest': config.get('cost_backtest', 0),
'portfolio_monitor': config.get('cost_portfolio_monitor', 0),
}
}
# 全局单例
_billing_service = None
def get_billing_service() -> BillingService:
"""获取计费服务单例"""
global _billing_service
if _billing_service is None:
_billing_service = BillingService()
return _billing_service
backend_api_python/app/services/email_service.py
+361
View File
@@ -0,0 +1,361 @@
"""
Email Service - Handles email verification codes and notifications.
"""
import os
import random
import string
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from datetime import datetime, timedelta
from typing import Tuple, Optional
from app.utils.db import get_db_connection
from app.utils.logger import get_logger
logger = get_logger(__name__)
# Singleton instance
_email_service = None
def get_email_service():
"""Get singleton EmailService instance"""
global _email_service
if _email_service is None:
_email_service = EmailService()
return _email_service
class EmailService:
"""Email service for verification codes and notifications"""
def __init__(self):
self._load_config()
def _load_config(self):
"""Load email configuration from environment variables"""
self.smtp_host = os.getenv('SMTP_HOST', '')
self.smtp_port = int(os.getenv('SMTP_PORT', '587'))
self.smtp_user = os.getenv('SMTP_USER', '')
self.smtp_password = os.getenv('SMTP_PASSWORD', '')
self.smtp_from = os.getenv('SMTP_FROM', '') or self.smtp_user
self.smtp_use_tls = os.getenv('SMTP_USE_TLS', 'true').lower() == 'true'
self.smtp_use_ssl = os.getenv('SMTP_USE_SSL', 'false').lower() == 'true'
# Verification code settings
self.code_expire_minutes = int(os.getenv('VERIFICATION_CODE_EXPIRE_MINUTES', '10'))
self.code_length = 6
# Verification code attempt limits (anti-brute-force)
self.code_max_attempts = int(os.getenv('VERIFICATION_CODE_MAX_ATTEMPTS', '5'))
self.code_lock_minutes = int(os.getenv('VERIFICATION_CODE_LOCK_MINUTES', '30'))
# Check if email is properly configured
self.email_enabled = bool(self.smtp_host and self.smtp_user and self.smtp_password)
if not self.email_enabled:
logger.warning("Email service is not configured. SMTP settings are missing.")
def is_configured(self) -> bool:
"""Check if email service is properly configured"""
return self.email_enabled
# =========================================================================
# Verification Code Generation & Storage
# =========================================================================
def generate_code(self) -> str:
"""Generate a random numeric verification code"""
return ''.join(random.choices(string.digits, k=self.code_length))
def create_verification_code(self, email: str, code_type: str,
ip_address: str = None) -> Tuple[bool, str]:
"""
Create and store a new verification code.
Args:
email: Email address
code_type: Type of verification (register, reset_password, change_password, change_email)
ip_address: Requester's IP address
Returns:
(success, code_or_message)
"""
try:
code = self.generate_code()
expires_at = datetime.now() + timedelta(minutes=self.code_expire_minutes)
with get_db_connection() as db:
cur = db.cursor()
# Invalidate any existing unused codes of the same type for this email
cur.execute(
"""
UPDATE qd_verification_codes
SET used_at = NOW()
WHERE email = ? AND type = ? AND used_at IS NULL
""",
(email, code_type)
)
# Insert new code
cur.execute(
"""
INSERT INTO qd_verification_codes
(email, code, type, expires_at, ip_address)
VALUES (?, ?, ?, ?, ?)
""",
(email, code, code_type, expires_at, ip_address)
)
db.commit()
cur.close()
return True, code
except Exception as e:
logger.error(f"Failed to create verification code: {e}")
return False, 'Failed to generate verification code'
def verify_code(self, email: str, code: str, code_type: str) -> Tuple[bool, str]:
"""
Verify a submitted code with brute-force protection.
Args:
email: Email address
code: The code to verify
code_type: Type of verification
Returns:
(valid, message)
"""
try:
with get_db_connection() as db:
cur = db.cursor()
# Check if locked due to too many failed attempts
lock_window = datetime.now() - timedelta(minutes=self.code_lock_minutes)
cur.execute(
"""
SELECT COUNT(*) as cnt FROM qd_verification_codes
WHERE email = ? AND type = ?
AND attempts >= ? AND last_attempt_at > ?
AND used_at IS NULL
""",
(email, code_type, self.code_max_attempts, lock_window.isoformat())
)
lock_row = cur.fetchone()
if lock_row and lock_row['cnt'] > 0:
cur.close()
return False, f'Too many failed attempts. Please try again in {self.code_lock_minutes} minutes'
# Find latest unused code for this email/type
cur.execute(
"""
SELECT id, code as stored_code, expires_at, attempts FROM qd_verification_codes
WHERE email = ? AND type = ? AND used_at IS NULL
ORDER BY created_at DESC
LIMIT 1
""",
(email, code_type)
)
row = cur.fetchone()
if not row:
cur.close()
return False, 'Invalid verification code'
code_id = row['id']
stored_code = row['stored_code']
attempts = row['attempts'] or 0
# Check if code matches
if stored_code != code:
# Increment attempt counter
new_attempts = attempts + 1
cur.execute(
"""
UPDATE qd_verification_codes
SET attempts = ?, last_attempt_at = NOW()
WHERE id = ?
""",
(new_attempts, code_id)
)
db.commit()
cur.close()
remaining = self.code_max_attempts - new_attempts
if remaining <= 0:
return False, f'Too many failed attempts. Please try again in {self.code_lock_minutes} minutes'
return False, f'Invalid verification code. {remaining} attempts remaining'
# Check expiration
expires_at = row['expires_at']
if isinstance(expires_at, str):
expires_at = datetime.fromisoformat(expires_at)
if datetime.now() > expires_at:
cur.close()
return False, 'Verification code has expired'
# Mark as used
cur.execute(
"UPDATE qd_verification_codes SET used_at = NOW() WHERE id = ?",
(code_id,)
)
db.commit()
cur.close()
return True, 'verified'
except Exception as e:
logger.error(f"Failed to verify code: {e}")
return False, 'Verification failed'
# =========================================================================
# Email Sending
# =========================================================================
def send_email(self, to_email: str, subject: str, html_body: str) -> Tuple[bool, str]:
"""
Send an email.
Args:
to_email: Recipient email address
subject: Email subject
html_body: HTML body content
Returns:
(success, message)
"""
if not self.email_enabled:
logger.warning(f"Email not sent (service disabled): {subject} to {to_email}")
return False, 'Email service is not configured'
try:
msg = MIMEMultipart('alternative')
msg['Subject'] = subject
msg['From'] = self.smtp_from
msg['To'] = to_email
# Plain text version (fallback)
text_body = html_body.replace('<br>', '\n').replace('<br/>', '\n')
# Simple HTML tag removal for plain text
import re
text_body = re.sub('<[^<]+?>', '', text_body)
part1 = MIMEText(text_body, 'plain', 'utf-8')
part2 = MIMEText(html_body, 'html', 'utf-8')
msg.attach(part1)
msg.attach(part2)
# Connect and send
if self.smtp_use_ssl:
server = smtplib.SMTP_SSL(self.smtp_host, self.smtp_port)
else:
server = smtplib.SMTP(self.smtp_host, self.smtp_port)
if self.smtp_use_tls:
server.starttls()
server.login(self.smtp_user, self.smtp_password)
server.sendmail(self.smtp_from, to_email, msg.as_string())
server.quit()
logger.info(f"Email sent successfully: {subject} to {to_email}")
return True, 'sent'
except smtplib.SMTPAuthenticationError as e:
logger.error(f"SMTP authentication failed: {e}")
return False, 'Email authentication failed'
except smtplib.SMTPException as e:
logger.error(f"SMTP error: {e}")
return False, 'Failed to send email'
except Exception as e:
logger.error(f"Email error: {e}")
return False, 'Failed to send email'
def send_verification_code(self, email: str, code_type: str,
ip_address: str = None) -> Tuple[bool, str]:
"""
Generate and send a verification code email.
Args:
email: Recipient email address
code_type: Type of verification (register, reset_password, change_password)
ip_address: Requester's IP address
Returns:
(success, message)
"""
# Generate code
success, code_or_msg = self.create_verification_code(email, code_type, ip_address)
if not success:
return False, code_or_msg
code = code_or_msg
# Prepare email content based on type
if code_type == 'register':
subject = 'QuantDinger - Verification Code for Registration'
action_text = 'complete your registration'
elif code_type == 'login':
subject = 'QuantDinger - Quick Login Verification Code'
action_text = 'log in to your account'
elif code_type == 'reset_password':
subject = 'QuantDinger - Password Reset Verification Code'
action_text = 'reset your password'
elif code_type == 'change_password':
subject = 'QuantDinger - Verification Code for Password Change'
action_text = 'change your password'
elif code_type == 'change_email':
subject = 'QuantDinger - Verification Code for Email Change'
action_text = 'change your email address'
else:
subject = 'QuantDinger - Verification Code'
action_text = 'complete the verification'
html_body = f"""
<div style="font-family: Arial, sans-serif; max-width: 600px; margin: 0 auto; padding: 20px;">
<div style="text-align: center; margin-bottom: 30px;">
<h1 style="color: #1890ff; margin: 0;">QuantDinger</h1>
<p style="color: #666; margin-top: 5px;">AI-Driven Quantitative Insights</p>
</div>
<div style="background: #f5f5f5; border-radius: 8px; padding: 30px; text-align: center;">
<p style="color: #333; font-size: 16px; margin: 0 0 20px 0;">
Your verification code to {action_text} is:
</p>
<div style="background: #fff; border: 2px solid #1890ff; border-radius: 8px; padding: 20px; display: inline-block;">
<span style="font-size: 32px; font-weight: bold; letter-spacing: 8px; color: #1890ff;">{code}</span>
</div>
<p style="color: #999; font-size: 14px; margin-top: 20px;">
This code will expire in {self.code_expire_minutes} minutes.
</p>
</div>
<div style="margin-top: 30px; padding: 20px; background: #fff8e6; border-radius: 8px;">
<p style="color: #d48806; font-size: 14px; margin: 0;">
<strong>Security Notice:</strong> If you did not request this code,
please ignore this email. Do not share this code with anyone.
</p>
</div>
<div style="text-align: center; margin-top: 30px; color: #999; font-size: 12px;">
<p>&copy; QuantDinger. All rights reserved.</p>
</div>
</div>
"""
# Send email
return self.send_email(email, subject, html_body)
# =========================================================================
# Email Validation
# =========================================================================
@staticmethod
def is_valid_email(email: str) -> bool:
"""Basic email format validation"""
import re
pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
return bool(re.match(pattern, email))
backend_api_python/app/services/oauth_service.py
+517
View File
@@ -0,0 +1,517 @@
"""
OAuth Service - Handles Google and GitHub OAuth authentication.
"""
import os
import secrets
import requests
from urllib.parse import urlencode
from datetime import datetime
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
_oauth_service = None
def get_oauth_service():
"""Get singleton OAuthService instance"""
global _oauth_service
if _oauth_service is None:
_oauth_service = OAuthService()
return _oauth_service
class OAuthService:
"""OAuth service for Google and GitHub authentication"""
def __init__(self):
self._load_config()
def _load_config(self):
"""Load OAuth configuration from environment variables"""
# Google OAuth
self.google_client_id = os.getenv('GOOGLE_CLIENT_ID', '')
self.google_client_secret = os.getenv('GOOGLE_CLIENT_SECRET', '')
self.google_redirect_uri = os.getenv('GOOGLE_REDIRECT_URI', '')
self.google_enabled = bool(self.google_client_id and self.google_client_secret)
# GitHub OAuth
self.github_client_id = os.getenv('GITHUB_CLIENT_ID', '')
self.github_client_secret = os.getenv('GITHUB_CLIENT_SECRET', '')
self.github_redirect_uri = os.getenv('GITHUB_REDIRECT_URI', '')
self.github_enabled = bool(self.github_client_id and self.github_client_secret)
# Frontend URL for redirect after OAuth
self.frontend_url = os.getenv('FRONTEND_URL', 'http://localhost:8080')
# State storage (in-memory for simplicity, could use Redis in production)
self._states = {}
# =========================================================================
# Google OAuth
# =========================================================================
def get_google_auth_url(self, state: str = None) -> Tuple[str, str]:
"""
Generate Google OAuth authorization URL.
Returns:
(auth_url, state)
"""
if not self.google_enabled:
return '', ''
state = state or secrets.token_urlsafe(32)
self._states[state] = {'provider': 'google', 'created_at': datetime.now()}
params = {
'client_id': self.google_client_id,
'redirect_uri': self.google_redirect_uri,
'response_type': 'code',
'scope': 'openid email profile',
'state': state,
'access_type': 'offline',
'prompt': 'select_account'
}
auth_url = f"https://accounts.google.com/o/oauth2/v2/auth?{urlencode(params)}"
return auth_url, state
def handle_google_callback(self, code: str, state: str) -> Tuple[bool, Dict[str, Any]]:
"""
Handle Google OAuth callback.
Args:
code: Authorization code from Google
state: State parameter for CSRF protection
Returns:
(success, user_info_or_error)
"""
# Validate state
if state not in self._states or self._states[state].get('provider') != 'google':
return False, {'error': 'Invalid state parameter'}
del self._states[state]
try:
# Exchange code for tokens
token_response = requests.post(
'https://oauth2.googleapis.com/token',
data={
'code': code,
'client_id': self.google_client_id,
'client_secret': self.google_client_secret,
'redirect_uri': self.google_redirect_uri,
'grant_type': 'authorization_code'
},
timeout=10
)
if token_response.status_code != 200:
logger.error(f"Google token exchange failed: {token_response.text}")
return False, {'error': 'Failed to exchange authorization code'}
tokens = token_response.json()
access_token = tokens.get('access_token')
# Get user info
user_response = requests.get(
'https://www.googleapis.com/oauth2/v2/userinfo',
headers={'Authorization': f'Bearer {access_token}'},
timeout=10
)
if user_response.status_code != 200:
logger.error(f"Google user info failed: {user_response.text}")
return False, {'error': 'Failed to get user information'}
user_info = user_response.json()
return True, {
'provider': 'google',
'provider_user_id': user_info.get('id'),
'email': user_info.get('email'),
'name': user_info.get('name'),
'avatar': user_info.get('picture'),
'access_token': access_token,
'refresh_token': tokens.get('refresh_token')
}
except requests.RequestException as e:
logger.error(f"Google OAuth error: {e}")
return False, {'error': 'OAuth service unavailable'}
# =========================================================================
# GitHub OAuth
# =========================================================================
def get_github_auth_url(self, state: str = None) -> Tuple[str, str]:
"""
Generate GitHub OAuth authorization URL.
Returns:
(auth_url, state)
"""
if not self.github_enabled:
return '', ''
state = state or secrets.token_urlsafe(32)
self._states[state] = {'provider': 'github', 'created_at': datetime.now()}
params = {
'client_id': self.github_client_id,
'redirect_uri': self.github_redirect_uri,
'scope': 'user:email read:user',
'state': state
}
auth_url = f"https://github.com/login/oauth/authorize?{urlencode(params)}"
return auth_url, state
def handle_github_callback(self, code: str, state: str) -> Tuple[bool, Dict[str, Any]]:
"""
Handle GitHub OAuth callback.
Args:
code: Authorization code from GitHub
state: State parameter for CSRF protection
Returns:
(success, user_info_or_error)
"""
# Validate state
if state not in self._states or self._states[state].get('provider') != 'github':
return False, {'error': 'Invalid state parameter'}
del self._states[state]
try:
# Exchange code for token
token_response = requests.post(
'https://github.com/login/oauth/access_token',
data={
'client_id': self.github_client_id,
'client_secret': self.github_client_secret,
'code': code,
'redirect_uri': self.github_redirect_uri
},
headers={'Accept': 'application/json'},
timeout=10
)
if token_response.status_code != 200:
logger.error(f"GitHub token exchange failed: {token_response.text}")
return False, {'error': 'Failed to exchange authorization code'}
tokens = token_response.json()
access_token = tokens.get('access_token')
if not access_token:
error = tokens.get('error_description', 'Unknown error')
logger.error(f"GitHub token error: {error}")
return False, {'error': error}
# Get user info
user_response = requests.get(
'https://api.github.com/user',
headers={
'Authorization': f'Bearer {access_token}',
'Accept': 'application/vnd.github.v3+json'
},
timeout=10
)
if user_response.status_code != 200:
logger.error(f"GitHub user info failed: {user_response.text}")
return False, {'error': 'Failed to get user information'}
user_info = user_response.json()
# Get user email (might be private)
email = user_info.get('email')
if not email:
email_response = requests.get(
'https://api.github.com/user/emails',
headers={
'Authorization': f'Bearer {access_token}',
'Accept': 'application/vnd.github.v3+json'
},
timeout=10
)
if email_response.status_code == 200:
emails = email_response.json()
# Find primary email
for e in emails:
if e.get('primary') and e.get('verified'):
email = e.get('email')
break
# Fallback to any verified email
if not email:
for e in emails:
if e.get('verified'):
email = e.get('email')
break
return True, {
'provider': 'github',
'provider_user_id': str(user_info.get('id')),
'email': email,
'name': user_info.get('name') or user_info.get('login'),
'avatar': user_info.get('avatar_url'),
'access_token': access_token,
'refresh_token': None # GitHub doesn't use refresh tokens
}
except requests.RequestException as e:
logger.error(f"GitHub OAuth error: {e}")
return False, {'error': 'OAuth service unavailable'}
# =========================================================================
# OAuth Link Management
# =========================================================================
def get_or_create_user_from_oauth(self, oauth_info: Dict[str, Any]) -> Tuple[bool, Dict[str, Any]]:
"""
Get existing user or create new user from OAuth info.
Args:
oauth_info: Dict with provider, provider_user_id, email, name, avatar, tokens
Returns:
(success, user_or_error)
"""
provider = oauth_info['provider']
provider_user_id = oauth_info['provider_user_id']
email = oauth_info.get('email')
name = oauth_info.get('name', '')
avatar = oauth_info.get('avatar', '/avatar2.jpg')
try:
with get_db_connection() as db:
cur = db.cursor()
# Check if OAuth link exists
cur.execute(
"""
SELECT user_id FROM qd_oauth_links
WHERE provider = ? AND provider_user_id = ?
""",
(provider, provider_user_id)
)
link = cur.fetchone()
if link:
# Existing OAuth link - get user
user_id = link['user_id']
cur.execute(
"""
SELECT id, username, email, nickname, avatar, status, role
FROM qd_users WHERE id = ?
""",
(user_id,)
)
user = cur.fetchone()
if user:
# Update OAuth tokens
cur.execute(
"""
UPDATE qd_oauth_links
SET access_token = ?, refresh_token = ?, updated_at = NOW()
WHERE provider = ? AND provider_user_id = ?
""",
(oauth_info.get('access_token'), oauth_info.get('refresh_token'),
provider, provider_user_id)
)
# Update last login
cur.execute(
"UPDATE qd_users SET last_login_at = NOW() WHERE id = ?",
(user_id,)
)
db.commit()
cur.close()
return True, dict(user)
else:
# Orphaned OAuth link - remove it
cur.execute(
"DELETE FROM qd_oauth_links WHERE provider = ? AND provider_user_id = ?",
(provider, provider_user_id)
)
db.commit()
# Check if user exists with same email
if email:
cur.execute(
"""
SELECT id, username, email, nickname, avatar, status, role
FROM qd_users WHERE email = ?
""",
(email,)
)
existing_user = cur.fetchone()
if existing_user:
# Link OAuth to existing user
cur.execute(
"""
INSERT INTO qd_oauth_links
(user_id, provider, provider_user_id, provider_email,
provider_name, provider_avatar, access_token, refresh_token)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
(existing_user['id'], provider, provider_user_id, email,
name, avatar, oauth_info.get('access_token'),
oauth_info.get('refresh_token'))
)
cur.execute(
"UPDATE qd_users SET last_login_at = NOW() WHERE id = ?",
(existing_user['id'],)
)
db.commit()
cur.close()
return True, dict(existing_user)
# Create new user
# Generate unique username from OAuth name or email
base_username = (name or email.split('@')[0] if email else provider_user_id)
base_username = ''.join(c for c in base_username if c.isalnum() or c in '_-')[:30]
username = base_username
# Ensure username is unique
counter = 1
while True:
cur.execute("SELECT id FROM qd_users WHERE username = ?", (username,))
if not cur.fetchone():
break
username = f"{base_username}_{counter}"
counter += 1
# Generate a random password (user won't need it for OAuth login)
import secrets
random_password = secrets.token_urlsafe(32)
from app.services.user_service import get_user_service
password_hash = get_user_service().hash_password(random_password)
# Ensure email is unique or generate placeholder
if email:
cur.execute("SELECT id FROM qd_users WHERE email = ?", (email,))
if cur.fetchone():
email = f"{provider}_{provider_user_id}@oauth.local"
else:
email = f"{provider}_{provider_user_id}@oauth.local"
# Insert new user
cur.execute(
"""
INSERT INTO qd_users
(username, password_hash, email, nickname, avatar, status, role, email_verified)
VALUES (?, ?, ?, ?, ?, 'active', 'user', TRUE)
""",
(username, password_hash, email, name or username, avatar or '/avatar2.jpg')
)
user_id = cur.lastrowid
# Create OAuth link
cur.execute(
"""
INSERT INTO qd_oauth_links
(user_id, provider, provider_user_id, provider_email,
provider_name, provider_avatar, access_token, refresh_token)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
(user_id, provider, provider_user_id, oauth_info.get('email'),
name, avatar, oauth_info.get('access_token'),
oauth_info.get('refresh_token'))
)
db.commit()
cur.close()
return True, {
'id': user_id,
'username': username,
'email': email,
'nickname': name or username,
'avatar': avatar or '/avatar2.jpg',
'status': 'active',
'role': 'user'
}
except Exception as e:
logger.error(f"OAuth user creation failed: {e}")
return False, {'error': 'Failed to create user account'}
def get_user_oauth_links(self, user_id: int) -> list:
"""Get all OAuth links for a user"""
try:
with get_db_connection() as db:
cur = db.cursor()
cur.execute(
"""
SELECT provider, provider_email, provider_name, created_at
FROM qd_oauth_links WHERE user_id = ?
""",
(user_id,)
)
links = cur.fetchall()
cur.close()
return [dict(link) for link in links] if links else []
except Exception as e:
logger.error(f"Failed to get OAuth links: {e}")
return []
def unlink_oauth(self, user_id: int, provider: str) -> Tuple[bool, str]:
"""Unlink an OAuth provider from user account"""
try:
with get_db_connection() as db:
cur = db.cursor()
# Check if user has password (can't unlink last auth method)
cur.execute(
"SELECT password_hash FROM qd_users WHERE id = ?",
(user_id,)
)
user = cur.fetchone()
if not user or not user['password_hash']:
# Check if this is the only OAuth link
cur.execute(
"SELECT COUNT(*) as count FROM qd_oauth_links WHERE user_id = ?",
(user_id,)
)
count = cur.fetchone()['count']
if count <= 1:
cur.close()
return False, 'Cannot unlink the only authentication method'
cur.execute(
"DELETE FROM qd_oauth_links WHERE user_id = ? AND provider = ?",
(user_id, provider)
)
db.commit()
cur.close()
return True, 'unlinked'
except Exception as e:
logger.error(f"Failed to unlink OAuth: {e}")
return False, 'Failed to unlink account'
# =========================================================================
# Cleanup
# =========================================================================
def cleanup_expired_states(self, max_age_minutes: int = 10):
"""Clean up expired OAuth states"""
cutoff = datetime.now()
from datetime import timedelta
cutoff = cutoff - timedelta(minutes=max_age_minutes)
expired = [k for k, v in self._states.items()
if v.get('created_at', datetime.now()) < cutoff]
for k in expired:
del self._states[k]
backend_api_python/app/services/security_service.py
+393
View File
@@ -0,0 +1,393 @@
"""
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
backend_api_python/app/services/user_service.py
+104 -32
View File
@@ -75,7 +75,7 @@ class UserService:
cur.execute(
"""
SELECT id, username, email, nickname, avatar, status, role,
last_login_at, created_at, updated_at
credits, vip_expires_at, last_login_at, created_at, updated_at
FROM qd_users WHERE id = ?
""",
(user_id,)
@@ -107,12 +107,41 @@ class UserService:
logger.error(f"get_user_by_username failed: {e}")
return None
def get_user_by_email(self, email: str) -> Optional[Dict[str, Any]]:
"""Get user by email (includes password_hash for auth)"""
if not email:
return None
try:
with get_db_connection() as db:
cur = db.cursor()
cur.execute(
"""
SELECT id, username, password_hash, email, nickname, avatar,
status, role, last_login_at, created_at, updated_at
FROM qd_users WHERE LOWER(email) = LOWER(?)
""",
(email,)
)
row = cur.fetchone()
cur.close()
return row
except Exception as e:
logger.error(f"get_user_by_email failed: {e}")
return None
def authenticate(self, username: str, password: str) -> Optional[Dict[str, Any]]:
"""
Authenticate user with username and password.
Authenticate user with username/email and password.
Supports both username and email login.
Returns user info (without password_hash) if successful, None otherwise.
"""
# Try username first
user = self.get_user_by_username(username)
# If not found, try email (supports both username and email login)
if not user:
user = self.get_user_by_email(username)
if not user:
return None
@@ -120,7 +149,16 @@ class UserService:
logger.warning(f"Login attempt for disabled user: {username}")
return None
if not self.verify_password(password, user.get('password_hash', '')):
password_hash = user.get('password_hash', '')
# Check if user has no password (code-login user)
if not password_hash or password_hash.strip() == '':
logger.info(f"Password login attempted for code-login user: {username}")
# Return a special marker to indicate no password set
# This allows the caller to provide a more specific error message
return {'_no_password': True, **user}
if not self.verify_password(password, password_hash):
return None
# Update last login time
@@ -140,33 +178,41 @@ class UserService:
user.pop('password_hash', None)
return user
def create_user(self, data: Dict[str, Any]) -> Optional[int]:
def create_user(self, data: Dict[str, Any] = None, **kwargs) -> Optional[int]:
"""
Create a new user.
Args:
data: {
data: dict with user fields, OR use keyword arguments:
username: str (required),
password: str (required),
password: str (optional, can be None for code-login users),
email: str (optional),
nickname: str (optional),
role: str (optional, default 'user'),
status: str (optional, default 'active')
}
status: str (optional, default 'active'),
email_verified: bool (optional, default False),
referred_by: int (optional, referrer user ID)
Returns:
New user ID or None if failed
"""
username = (data.get('username') or '').strip()
password = data.get('password') or ''
# Support both dict and kwargs style
if data is None:
data = kwargs
else:
data = {**data, **kwargs}
if not username or not password:
raise ValueError("Username and password are required")
username = (data.get('username') or '').strip()
password = data.get('password') # Can be None for code-login users
if not username:
raise ValueError("Username is required")
if len(username) < 3 or len(username) > 50:
raise ValueError("Username must be 3-50 characters")
if len(password) < 6:
# Password validation only if provided
if password and len(password) < 6:
raise ValueError("Password must be at least 6 characters")
# Check if username already exists
@@ -174,11 +220,14 @@ class UserService:
if existing:
raise ValueError("Username already exists")
password_hash = self.hash_password(password)
# Hash password or use empty string for code-login users
password_hash = self.hash_password(password) if password else ''
email = (data.get('email') or '').strip() or None
nickname = (data.get('nickname') or '').strip() or username
role = data.get('role', 'user')
status = data.get('status', 'active')
email_verified = data.get('email_verified', False)
referred_by = data.get('referred_by') # Referrer user ID
if role not in self.ROLES:
role = 'user'
@@ -189,10 +238,10 @@ class UserService:
cur.execute(
"""
INSERT INTO qd_users
(username, password_hash, email, nickname, role, status, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, NOW(), NOW())
(username, password_hash, email, nickname, role, status, email_verified, referred_by, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, NOW(), NOW())
""",
(username, password_hash, email, nickname, role, status)
(username, password_hash, email, nickname, role, status, email_verified, referred_by)
)
db.commit()
user_id = cur.lastrowid
@@ -206,7 +255,7 @@ class UserService:
user_id = row['id'] if row else None
cur.close()
logger.info(f"Created user: {username} (id={user_id})")
logger.info(f"Created user: {username} (id={user_id}, referred_by={referred_by})")
return user_id
except Exception as e:
logger.error(f"create_user failed: {e}")
@@ -251,7 +300,7 @@ class UserService:
return False
def change_password(self, user_id: int, old_password: str, new_password: str) -> bool:
"""Change user password (requires old password verification)"""
"""Change user password (requires old password verification, except for users with no password)"""
user = self.get_user_by_id(user_id)
if not user:
return False
@@ -266,7 +315,15 @@ class UserService:
if not row:
return False
if not self.verify_password(old_password, row['password_hash']):
password_hash = row.get('password_hash', '')
# If user has no password (code-login user), allow setting password without old password
if not password_hash or password_hash.strip() == '':
logger.info(f"Setting initial password for code-login user: {user_id}")
return self.reset_password(user_id, new_password)
# For users with existing password, verify old password
if not self.verify_password(old_password, password_hash):
return False
return self.reset_password(user_id, new_password)
@@ -292,6 +349,10 @@ class UserService:
logger.error(f"reset_password failed: {e}")
return False
def update_password(self, user_id: int, new_password: str) -> bool:
"""Alias for reset_password - update user password without old password verification"""
return self.reset_password(user_id, new_password)
def delete_user(self, user_id: int) -> bool:
"""Delete a user"""
try:
@@ -305,29 +366,37 @@ class UserService:
logger.error(f"delete_user failed: {e}")
return False
def list_users(self, page: int = 1, page_size: int = 20) -> Dict[str, Any]:
"""List all users with pagination"""
def list_users(self, page: int = 1, page_size: int = 20, search: str = None) -> Dict[str, Any]:
"""List all users with pagination and optional search"""
offset = (page - 1) * page_size
try:
with get_db_connection() as db:
cur = db.cursor()
# Build WHERE clause for search
where_clause = ""
params = []
if search and search.strip():
search_term = f"%{search.strip()}%"
where_clause = "WHERE username LIKE ? OR email LIKE ? OR nickname LIKE ?"
params = [search_term, search_term, search_term]
# Get total count
cur.execute("SELECT COUNT(*) as count FROM qd_users")
count_sql = f"SELECT COUNT(*) as count FROM qd_users {where_clause}"
cur.execute(count_sql, tuple(params))
total = cur.fetchone()['count']
# Get users
cur.execute(
"""
query_sql = f"""
SELECT id, username, email, nickname, avatar, status, role,
last_login_at, created_at, updated_at
credits, vip_expires_at, last_login_at, created_at, updated_at
FROM qd_users
{where_clause}
ORDER BY id DESC
LIMIT ? OFFSET ?
""",
(page_size, offset)
)
"""
cur.execute(query_sql, tuple(params + [page_size, offset]))
users = cur.fetchall()
cur.close()
@@ -362,15 +431,18 @@ class UserService:
# Create admin using env credentials
admin_user = os.getenv('ADMIN_USER', 'admin')
admin_password = os.getenv('ADMIN_PASSWORD', 'admin123')
admin_email = os.getenv('ADMIN_EMAIL', 'admin@example.com')
self.create_user({
'username': admin_user,
'password': admin_password,
'email': admin_email,
'nickname': 'Administrator',
'role': 'admin',
'status': 'active'
'status': 'active',
'email_verified': True # Admin email is pre-verified
})
logger.info(f"Created admin user: {admin_user}")
logger.info(f"Created admin user: {admin_user} ({admin_email})")
except Exception as e:
logger.error(f"ensure_admin_exists failed: {e}")