TIANHE
1087 lines
43 KiB
Python
1087 lines
43 KiB
Python
"""
|
|
Authentication API Routes
|
|
|
|
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, redirect
|
|
from urllib.parse import urlencode
|
|
from app.config.settings import Config
|
|
from app.utils.auth import generate_token, login_required, authenticate_legacy
|
|
from app.utils.logger import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
auth_bp = Blueprint('auth', __name__)
|
|
|
|
def _build_frontend_login_redirect(frontend_url: str, **params) -> str:
|
|
"""
|
|
Build a redirect URL to frontend login page for OAuth flows.
|
|
|
|
Frontend uses Vue Router hash mode (`/#/user/login`), so redirecting to `/user/login`
|
|
will 404 on static hosting. Always normalize to `{origin}/#/user/login`.
|
|
"""
|
|
base = (frontend_url or '').strip().rstrip('/')
|
|
if not base:
|
|
base = 'http://localhost:8080'
|
|
|
|
if '/#/' in base:
|
|
origin = base.split('/#/', 1)[0].rstrip('/')
|
|
elif '#' in base:
|
|
origin = base.split('#', 1)[0].rstrip('/')
|
|
else:
|
|
origin = base
|
|
|
|
login_url = f"{origin}/#/user/login"
|
|
qs = urlencode({k: v for k, v in params.items() if v is not None and v != ''})
|
|
return f"{login_url}?{qs}" if qs else login_url
|
|
|
|
|
|
def _is_single_user_mode() -> bool:
|
|
"""Check if system is in single-user (legacy) mode"""
|
|
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():
|
|
"""
|
|
User login endpoint.
|
|
|
|
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/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
|
|
|
|
# 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}")
|
|
|
|
# Fallback to legacy single-user mode
|
|
if not user:
|
|
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
|
|
|
|
# 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: Increment token_version (invalidates old sessions for single-client login)
|
|
user_id = user.get('id') or user.get('user_id', 1)
|
|
try:
|
|
from app.services.user_service import get_user_service
|
|
new_token_version = get_user_service().increment_token_version(user_id)
|
|
except Exception as e:
|
|
logger.warning(f"Failed to increment token_version: {e}")
|
|
new_token_version = 1
|
|
|
|
# Step 5: Generate token with new token_version
|
|
token = generate_token(
|
|
user_id=user_id,
|
|
username=user.get('username', username),
|
|
role=user.get('role', 'admin'),
|
|
token_version=new_token_version # 包含新的 token_version
|
|
)
|
|
|
|
if not token:
|
|
return jsonify({'code': 500, 'msg': 'Token generation error', 'data': None}), 500
|
|
|
|
# Step 6: 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),
|
|
'username': user.get('username', username),
|
|
'nickname': user.get('nickname', 'User') + (' (Demo)' if is_demo else ''),
|
|
'avatar': user.get('avatar', '/avatar2.jpg'),
|
|
'is_demo': is_demo,
|
|
'role': {
|
|
'id': user.get('role', 'admin'),
|
|
'permissions': _get_permissions(user.get('role', 'admin'))
|
|
}
|
|
}
|
|
|
|
return jsonify({
|
|
'code': 1,
|
|
'msg': 'Login successful',
|
|
'data': {
|
|
'token': token,
|
|
'userinfo': userinfo
|
|
}
|
|
})
|
|
|
|
except Exception as e:
|
|
logger.error(f"Login error: {e}")
|
|
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
|
|
|
|
# Increment token_version (invalidates old sessions for single-client login)
|
|
try:
|
|
new_token_version = user_service.increment_token_version(user['id'])
|
|
except Exception as e:
|
|
logger.warning(f"Failed to increment token_version: {e}")
|
|
new_token_version = 1
|
|
|
|
# Generate token with new token_version
|
|
token = generate_token(
|
|
user_id=user['id'],
|
|
username=user['username'],
|
|
role=user.get('role', 'user'),
|
|
token_version=new_token_version
|
|
)
|
|
|
|
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()
|
|
affected = cur.rowcount
|
|
cur.close()
|
|
if affected == 0:
|
|
logger.error(f"Failed to update last_login_at: no rows affected for user_id={user['id']}")
|
|
else:
|
|
logger.info(f"Updated last_login_at for user_id={user['id']}")
|
|
except Exception as e:
|
|
logger.error(f"Failed to update last_login_at for user_id={user.get('id')}: {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 (get token_version for new user)
|
|
try:
|
|
new_token_version = user_service.get_token_version(user_id)
|
|
except Exception as e:
|
|
logger.warning(f"Failed to get token_version: {e}")
|
|
new_token_version = 1
|
|
|
|
token = generate_token(
|
|
user_id=user_id,
|
|
username=username,
|
|
role='user',
|
|
token_version=new_token_version
|
|
)
|
|
|
|
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(_build_frontend_login_redirect(frontend_url, oauth_error=error))
|
|
|
|
if not code or not state:
|
|
return redirect(_build_frontend_login_redirect(frontend_url, 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(_build_frontend_login_redirect(frontend_url, 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(_build_frontend_login_redirect(frontend_url, oauth_error=error_msg))
|
|
|
|
# Increment token_version (invalidates old sessions for single-client login)
|
|
from app.services.user_service import get_user_service
|
|
user_service = get_user_service()
|
|
try:
|
|
new_token_version = user_service.increment_token_version(user_result['id'])
|
|
except Exception as e:
|
|
logger.warning(f"Failed to increment token_version: {e}")
|
|
new_token_version = 1
|
|
|
|
# Generate token with new token_version
|
|
token = generate_token(
|
|
user_id=user_result['id'],
|
|
username=user_result['username'],
|
|
role=user_result.get('role', 'user'),
|
|
token_version=new_token_version
|
|
)
|
|
|
|
# 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(_build_frontend_login_redirect(frontend_url, 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(_build_frontend_login_redirect(frontend_url, 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(_build_frontend_login_redirect(frontend_url, oauth_error=error))
|
|
|
|
if not code or not state:
|
|
return redirect(_build_frontend_login_redirect(frontend_url, 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(_build_frontend_login_redirect(frontend_url, 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(_build_frontend_login_redirect(frontend_url, oauth_error=error_msg))
|
|
|
|
# Increment token_version (invalidates old sessions for single-client login)
|
|
from app.services.user_service import get_user_service
|
|
user_service = get_user_service()
|
|
try:
|
|
new_token_version = user_service.increment_token_version(user_result['id'])
|
|
except Exception as e:
|
|
logger.warning(f"Failed to increment token_version: {e}")
|
|
new_token_version = 1
|
|
|
|
# Generate token with new token_version
|
|
token = generate_token(
|
|
user_id=user_result['id'],
|
|
username=user_result['username'],
|
|
role=user_result.get('role', 'user'),
|
|
token_version=new_token_version
|
|
)
|
|
|
|
# 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(_build_frontend_login_redirect(frontend_url, 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(_build_frontend_login_redirect(frontend_url, oauth_error='server_error'))
|
|
|
|
|
|
# =============================================================================
|
|
# Other Endpoints
|
|
# =============================================================================
|
|
|
|
@auth_bp.route('/logout', methods=['POST'])
|
|
def logout():
|
|
"""Logout (client removes token; server is stateless)."""
|
|
return jsonify({'code': 1, 'msg': 'Logout successful', 'data': None})
|
|
|
|
|
|
@auth_bp.route('/info', methods=['GET'])
|
|
@login_required
|
|
def get_user_info():
|
|
"""Get current user info."""
|
|
try:
|
|
is_demo = os.getenv('IS_DEMO_MODE', 'false').lower() == 'true'
|
|
|
|
user_id = getattr(g, 'user_id', 1)
|
|
username = getattr(g, 'user', Config.ADMIN_USER)
|
|
role = getattr(g, 'user_role', 'admin')
|
|
|
|
# Try to get full user info from database
|
|
user_data = None
|
|
if not _is_single_user_mode():
|
|
try:
|
|
from app.services.user_service import get_user_service
|
|
user_data = get_user_service().get_user_by_id(user_id)
|
|
except Exception as e:
|
|
logger.warning(f"Failed to get user from database: {e}")
|
|
|
|
if user_data:
|
|
return jsonify({
|
|
'code': 1,
|
|
'msg': 'Success',
|
|
'data': {
|
|
'id': user_data.get('id'),
|
|
'username': user_data.get('username'),
|
|
'nickname': user_data.get('nickname', 'User') + (' (Demo)' if is_demo else ''),
|
|
'email': user_data.get('email'),
|
|
'avatar': user_data.get('avatar', '/avatar2.jpg'),
|
|
'is_demo': is_demo,
|
|
'role': {
|
|
'id': user_data.get('role', 'user'),
|
|
'permissions': _get_permissions(user_data.get('role', 'user'))
|
|
}
|
|
}
|
|
})
|
|
|
|
# Fallback for legacy mode
|
|
return jsonify({
|
|
'code': 1,
|
|
'msg': 'Success',
|
|
'data': {
|
|
'id': user_id,
|
|
'username': username,
|
|
'nickname': 'Admin' + (' (Demo)' if is_demo else ''),
|
|
'avatar': '/avatar2.jpg',
|
|
'is_demo': is_demo,
|
|
'role': {
|
|
'id': role,
|
|
'permissions': _get_permissions(role)
|
|
}
|
|
}
|
|
})
|
|
except Exception as e:
|
|
logger.error(f"get_user_info error: {e}")
|
|
return jsonify({'code': 500, 'msg': str(e), 'data': None}), 500
|
|
|
|
|
|
def _get_permissions(role: str) -> list:
|
|
"""Get permissions list for a role"""
|
|
try:
|
|
from app.services.user_service import get_user_service
|
|
return get_user_service().get_user_permissions(role)
|
|
except Exception:
|
|
# Default permissions for admin
|
|
if role == 'admin':
|
|
return ['dashboard', 'view', 'indicator', 'backtest', 'strategy',
|
|
'portfolio', 'settings', 'user_manage', 'credentials']
|
|
return ['dashboard', 'view', 'indicator', 'backtest', 'strategy', 'portfolio']
|