473 lines
17 KiB
Python
473 lines
17 KiB
Python
"""Authentication routes for user login, logout, and registration.
|
|
|
|
This module handles user authentication including login with account lockout
|
|
protection, logout with session clearing, and new user registration with
|
|
password policy enforcement and CAPTCHA verification.
|
|
"""
|
|
|
|
import uuid
|
|
import os
|
|
from datetime import datetime, timedelta
|
|
from flask import Blueprint, render_template, redirect, url_for, flash, request, session
|
|
from flask_login import login_user, logout_user, login_required, current_user
|
|
from app.extensions import db, hash_password, check_password, limiter
|
|
from app.models import User, Player, ESPORT_GAMES
|
|
from app.validators import RegisterSchema, LoginSchema
|
|
from marshmallow import ValidationError
|
|
from urllib.parse import urlparse
|
|
import requests
|
|
|
|
# Account lockout settings
|
|
MAX_LOGIN_ATTEMPTS = 5
|
|
LOCKOUT_DURATION_MINUTES = 15
|
|
|
|
# Discord OAuth2 configuration
|
|
DISCORD_CLIENT_ID = os.getenv('DISCORD_CLIENT_ID')
|
|
DISCORD_CLIENT_SECRET = os.getenv('DISCORD_CLIENT_SECRET')
|
|
DISCORD_REDIRECT_URI = os.getenv('DISCORD_REDIRECT_URI')
|
|
DISCORD_API_BASE = 'https://discord.com/api/v10'
|
|
|
|
# Mapping from Discord connection platform to E-Sports games
|
|
DISCORD_PLATFORM_TO_GAMES = {
|
|
'steam': ['Counter-Strike 2'],
|
|
'battlenet': ['Overwatch 2'],
|
|
'epicgames': ['Rocket League'],
|
|
'xbox': ['Apex Legends', 'Rainbow Six Siege', 'Rocket League'],
|
|
'playstation': ['Apex Legends', 'Rainbow Six Siege', 'Rocket League'],
|
|
}
|
|
|
|
|
|
def is_safe_url(url):
|
|
"""Validate that a URL is safe for redirection (same origin).
|
|
|
|
Args:
|
|
url: The URL to validate.
|
|
|
|
Returns:
|
|
bool: True if the URL is safe (relative or same origin).
|
|
"""
|
|
if not url:
|
|
return False
|
|
parsed = urlparse(url)
|
|
# Allow relative URLs (no netloc) or same-origin URLs
|
|
return not parsed.netloc or parsed.netloc == request.host
|
|
|
|
|
|
def generate_captcha():
|
|
"""Generate a simple math CAPTCHA challenge.
|
|
|
|
Creates a random addition problem and stores the answer in the session.
|
|
|
|
Returns:
|
|
dict: A dictionary with 'question' (e.g., '3 + 7') and 'id' keys.
|
|
"""
|
|
import random
|
|
a = random.randint(1, 10)
|
|
b = random.randint(1, 10)
|
|
captcha_id = str(uuid.uuid4())
|
|
session['captcha_id'] = captcha_id
|
|
session['captcha_answer'] = a + b
|
|
return {'question': f'{a} + {b} = ?', 'id': captcha_id}
|
|
|
|
|
|
def verify_captcha(user_answer):
|
|
"""Verify the CAPTCHA answer from the session.
|
|
|
|
Args:
|
|
user_answer: The user's submitted answer (string or int).
|
|
|
|
Returns:
|
|
bool: True if the answer matches the stored CAPTCHA, False otherwise.
|
|
"""
|
|
try:
|
|
expected = session.pop('captcha_answer', None)
|
|
session.pop('captcha_id', None)
|
|
if expected is None:
|
|
return False
|
|
return int(user_answer) == expected
|
|
except (ValueError, TypeError):
|
|
return False
|
|
|
|
|
|
auth_bp = Blueprint('auth', __name__, url_prefix='/auth')
|
|
|
|
|
|
@auth_bp.route('/login', methods=['GET', 'POST'])
|
|
@limiter.limit("10 per minute")
|
|
def login():
|
|
"""Handle user login authentication with account lockout protection.
|
|
|
|
GET: Render the login form.
|
|
POST: Authenticate user credentials with lockout check and audit logging.
|
|
|
|
Account lockout: After 5 consecutive failed attempts, the account is
|
|
locked for 15 minutes. Successful login resets the counter.
|
|
|
|
Redirects authenticated users to dashboard. Validates credentials and checks
|
|
account status before login.
|
|
|
|
Returns:
|
|
Response: Login form or redirect to dashboard/next page.
|
|
"""
|
|
if current_user.is_authenticated:
|
|
return redirect(url_for('main.dashboard'))
|
|
|
|
if request.method == 'POST':
|
|
# Validate input with marshmallow schema
|
|
login_schema = LoginSchema()
|
|
try:
|
|
validated = login_schema.load(request.form)
|
|
except ValidationError as err:
|
|
for field, messages in err.messages.items():
|
|
for msg in messages:
|
|
flash(f'{field}: {msg}', 'danger')
|
|
return render_template('pages/login.html')
|
|
|
|
username = validated['username']
|
|
password = validated['password']
|
|
user = User.query.filter_by(username=username).first()
|
|
|
|
# Check if account is locked
|
|
if user and user.locked_until and user.locked_until > datetime.utcnow():
|
|
remaining = (user.locked_until - datetime.utcnow()).seconds // 60
|
|
flash(
|
|
f'Account is locked due to too many failed attempts. '
|
|
f'Please try again in {remaining} minute(s).',
|
|
'danger'
|
|
)
|
|
return render_template('pages/login.html')
|
|
|
|
if user and check_password(user.password_hash, password):
|
|
if not user.is_active_account:
|
|
flash('This account has been deactivated.', 'danger')
|
|
return render_template('pages/login.html')
|
|
|
|
# Reset failed login attempts on successful login
|
|
user.failed_login_attempts = 0
|
|
user.locked_until = None
|
|
db.session.commit()
|
|
|
|
# Clear old session data and preserve CSRF token to prevent
|
|
# session fixation attacks (Flask-Login rotates the session ID)
|
|
_csrf_token = session.get('csrf_token')
|
|
session.clear()
|
|
if _csrf_token:
|
|
session['csrf_token'] = _csrf_token
|
|
|
|
login_user(user)
|
|
|
|
# Validate redirect URL to prevent open redirect vulnerability
|
|
next_page = request.args.get('next')
|
|
if next_page and not is_safe_url(next_page):
|
|
next_page = None
|
|
flash(f'Welcome back, {user.username}!', 'success')
|
|
return redirect(next_page) if next_page else redirect(url_for('main.dashboard'))
|
|
else:
|
|
# Track failed login attempt
|
|
if user:
|
|
user.failed_login_attempts += 1
|
|
if user.failed_login_attempts >= MAX_LOGIN_ATTEMPTS:
|
|
user.locked_until = datetime.utcnow() + timedelta(minutes=LOCKOUT_DURATION_MINUTES)
|
|
flash(
|
|
f'Account locked after {MAX_LOGIN_ATTEMPTS} failed attempts. '
|
|
f'Please try again in {LOCKOUT_DURATION_MINUTES} minutes.',
|
|
'danger'
|
|
)
|
|
else:
|
|
remaining = MAX_LOGIN_ATTEMPTS - user.failed_login_attempts
|
|
flash(
|
|
f'Login unsuccessful. {remaining} attempt(s) remaining before lockout.',
|
|
'danger'
|
|
)
|
|
db.session.commit()
|
|
else:
|
|
flash('Login unsuccessful. Please check username and password.', 'danger')
|
|
|
|
return render_template('pages/login.html')
|
|
|
|
|
|
@auth_bp.route('/register', methods=['GET', 'POST'])
|
|
@limiter.limit("20 per hour")
|
|
def register():
|
|
"""Handle new player registration with CAPTCHA and password policy.
|
|
|
|
GET: Render the registration form with E-Sports games list and CAPTCHA.
|
|
POST: Validate all inputs, verify CAPTCHA, enforce password policy,
|
|
and create a new player account.
|
|
|
|
Only players can register through this form. Validates username/email
|
|
uniqueness and password confirmation.
|
|
|
|
Returns:
|
|
Response: Registration form or redirect to login.
|
|
"""
|
|
if current_user.is_authenticated:
|
|
return redirect(url_for('main.dashboard'))
|
|
|
|
if request.method == 'POST':
|
|
# Build form data from request to preserve state across re-renders
|
|
form_data = dict(request.form)
|
|
form_data['games'] = request.form.getlist('games')
|
|
|
|
# Validate CAPTCHA first
|
|
captcha_answer = request.form.get('captcha_answer', '')
|
|
if not verify_captcha(captcha_answer):
|
|
flash('Incorrect CAPTCHA answer. Please try again.', 'danger')
|
|
captcha = generate_captcha()
|
|
# Clear password fields only on CAPTCHA failure
|
|
form_data.pop('password', None)
|
|
form_data.pop('confirm_password', None)
|
|
return render_template(
|
|
'pages/register.html',
|
|
esport_games=ESPORT_GAMES,
|
|
captcha=captcha,
|
|
form_data=form_data,
|
|
)
|
|
|
|
# Validate input with marshmallow schema
|
|
register_schema = RegisterSchema()
|
|
try:
|
|
validated = register_schema.load(form_data)
|
|
except ValidationError as err:
|
|
for field, messages in err.messages.items():
|
|
for msg in messages:
|
|
flash(f'{field}: {msg}', 'danger')
|
|
captcha = generate_captcha()
|
|
# Clear password fields on validation failure
|
|
form_data.pop('password', None)
|
|
form_data.pop('confirm_password', None)
|
|
return render_template(
|
|
'pages/register.html',
|
|
esport_games=ESPORT_GAMES,
|
|
captcha=captcha,
|
|
form_data=form_data,
|
|
)
|
|
|
|
username = validated['username']
|
|
email = validated['email']
|
|
password = validated['password']
|
|
full_name = validated['full_name']
|
|
phone = validated.get('phone')
|
|
selected_games = validated.get('games', [])
|
|
discord_username = validated.get('discord_username')
|
|
discord_user_id = validated.get('discord_user_id')
|
|
league_os_profile = validated.get('league_os_profile')
|
|
|
|
if User.query.filter_by(username=username).first():
|
|
flash('Username already exists.', 'danger')
|
|
captcha = generate_captcha()
|
|
form_data.pop('password', None)
|
|
form_data.pop('confirm_password', None)
|
|
return render_template(
|
|
'pages/register.html',
|
|
esport_games=ESPORT_GAMES,
|
|
captcha=captcha,
|
|
form_data=form_data,
|
|
)
|
|
|
|
if User.query.filter_by(email=email).first():
|
|
flash('Email already registered.', 'danger')
|
|
captcha = generate_captcha()
|
|
form_data.pop('password', None)
|
|
form_data.pop('confirm_password', None)
|
|
return render_template(
|
|
'pages/register.html',
|
|
esport_games=ESPORT_GAMES,
|
|
captcha=captcha,
|
|
form_data=form_data,
|
|
)
|
|
|
|
hashed_password = hash_password(password)
|
|
user = Player(
|
|
username=username,
|
|
password_hash=hashed_password,
|
|
role='player',
|
|
full_name=full_name,
|
|
email=email,
|
|
phone=phone,
|
|
games=','.join(selected_games) if selected_games else None,
|
|
discord_username=discord_username,
|
|
discord_user_id=discord_user_id,
|
|
league_os_profile=league_os_profile,
|
|
)
|
|
db.session.add(user)
|
|
db.session.commit()
|
|
|
|
# Create UserGamertag records for each selected game
|
|
from app.models import UserGamertag
|
|
for game in selected_games:
|
|
field_name = f'gamertag_{game}'
|
|
gamertag_value = request.form.get(field_name, '').strip()
|
|
if gamertag_value:
|
|
gamertag = UserGamertag(
|
|
user_id=user.id,
|
|
game=game,
|
|
gamertag=gamertag_value,
|
|
)
|
|
db.session.add(gamertag)
|
|
db.session.commit()
|
|
|
|
# Clear Discord OAuth data from session after successful registration
|
|
session.pop('discord_oauth', None)
|
|
|
|
flash('Your account has been created! You can now log in.', 'success')
|
|
return redirect(url_for('auth.login'))
|
|
|
|
# GET request — render empty form
|
|
captcha = generate_captcha()
|
|
return render_template(
|
|
'pages/register.html',
|
|
esport_games=ESPORT_GAMES,
|
|
captcha=captcha,
|
|
form_data={},
|
|
)
|
|
|
|
|
|
@auth_bp.route('/discord/login')
|
|
def discord_login():
|
|
"""Redirect the user to Discord's OAuth2 authorization page.
|
|
|
|
Requests the 'identify' and 'connections' scopes so we can retrieve
|
|
the user's Discord username, ID, and linked gaming accounts.
|
|
|
|
Returns:
|
|
Response: Redirect to Discord authorization URL.
|
|
"""
|
|
if not DISCORD_CLIENT_ID:
|
|
flash('Discord OAuth2 is not configured.', 'danger')
|
|
return redirect(url_for('auth.register'))
|
|
|
|
params = {
|
|
'client_id': DISCORD_CLIENT_ID,
|
|
'redirect_uri': DISCORD_REDIRECT_URI,
|
|
'response_type': 'code',
|
|
'scope': 'identify connections',
|
|
}
|
|
query = '&'.join(f'{k}={requests.utils.quote(v)}' for k, v in params.items())
|
|
auth_url = f'{DISCORD_API_BASE}/oauth2/authorize?{query}'
|
|
return redirect(auth_url)
|
|
|
|
|
|
@auth_bp.route('/discord/callback')
|
|
def discord_callback():
|
|
"""Handle the OAuth2 callback from Discord.
|
|
|
|
Exchanges the authorization code for an access token, then fetches
|
|
the user's profile (/users/@me) and connections (/users/@me/connections).
|
|
Results are stored in the session and the user is redirected back to
|
|
the registration form where fields will be pre-filled.
|
|
|
|
Returns:
|
|
Response: Redirect to registration page.
|
|
"""
|
|
code = request.args.get('code')
|
|
if not code:
|
|
flash('Discord authorization failed. No code received.', 'danger')
|
|
return redirect(url_for('auth.register'))
|
|
|
|
# Exchange the authorization code for an access token
|
|
token_data = {
|
|
'client_id': DISCORD_CLIENT_ID,
|
|
'client_secret': DISCORD_CLIENT_SECRET,
|
|
'grant_type': 'authorization_code',
|
|
'code': code,
|
|
'redirect_uri': DISCORD_REDIRECT_URI,
|
|
}
|
|
headers = {'Content-Type': 'application/x-www-form-urlencoded'}
|
|
|
|
try:
|
|
token_response = requests.post(
|
|
f'{DISCORD_API_BASE}/oauth2/token',
|
|
data=token_data,
|
|
headers=headers,
|
|
timeout=10,
|
|
)
|
|
token_response.raise_for_status()
|
|
token_json = token_response.json()
|
|
access_token = token_json.get('access_token')
|
|
except requests.RequestException as e:
|
|
flash(f'Failed to connect to Discord. Please try again.', 'danger')
|
|
return redirect(url_for('auth.register'))
|
|
|
|
if not access_token:
|
|
flash('Failed to obtain Discord access token.', 'danger')
|
|
return redirect(url_for('auth.register'))
|
|
|
|
auth_headers = {'Authorization': f'Bearer {access_token}'}
|
|
|
|
# Fetch the user's Discord profile
|
|
try:
|
|
user_response = requests.get(
|
|
f'{DISCORD_API_BASE}/users/@me',
|
|
headers=auth_headers,
|
|
timeout=10,
|
|
)
|
|
user_response.raise_for_status()
|
|
user_data = user_response.json()
|
|
except requests.RequestException:
|
|
flash('Failed to fetch Discord user profile.', 'danger')
|
|
return redirect(url_for('auth.register'))
|
|
|
|
# Fetch the user's connected gaming accounts
|
|
connections = []
|
|
try:
|
|
conn_response = requests.get(
|
|
f'{DISCORD_API_BASE}/users/@me/connections',
|
|
headers=auth_headers,
|
|
timeout=10,
|
|
)
|
|
conn_response.raise_for_status()
|
|
connections = conn_response.json()
|
|
except requests.RequestException:
|
|
# Non-critical: we can still proceed without connections
|
|
pass
|
|
|
|
# Build gamertag suggestions from Discord connections
|
|
gamertag_suggestions = {}
|
|
for conn in connections:
|
|
platform = conn.get('type', '')
|
|
name = conn.get('name', '').strip()
|
|
if not name or platform not in DISCORD_PLATFORM_TO_GAMES:
|
|
continue
|
|
for game in DISCORD_PLATFORM_TO_GAMES[platform]:
|
|
# Only set if not already set (first connection wins)
|
|
if game not in gamertag_suggestions:
|
|
gamertag_suggestions[game] = name
|
|
|
|
# Build a list of games to auto-select (unambiguous platform mappings)
|
|
auto_select_games = []
|
|
for conn in connections:
|
|
platform = conn.get('type', '')
|
|
if platform in ('steam', 'battlenet', 'epicgames'):
|
|
for game in DISCORD_PLATFORM_TO_GAMES[platform]:
|
|
if game not in auto_select_games:
|
|
auto_select_games.append(game)
|
|
|
|
# Store in session for the registration form to use
|
|
session['discord_oauth'] = {
|
|
'id': user_data.get('id'),
|
|
'username': user_data.get('username'),
|
|
'avatar': user_data.get('avatar'),
|
|
'gamertag_suggestions': gamertag_suggestions,
|
|
'auto_select_games': auto_select_games,
|
|
}
|
|
|
|
flash('Discord account connected! Your profile has been pre-filled.', 'success')
|
|
return redirect(url_for('auth.register'))
|
|
|
|
|
|
@auth_bp.route('/logout')
|
|
@login_required
|
|
def logout():
|
|
"""Log out the current user and clear the session.
|
|
|
|
Clears the user session and regenerates session ID to prevent
|
|
session fixation/replay after logout.
|
|
|
|
Returns:
|
|
Response: Redirect to login page with logout message.
|
|
"""
|
|
logout_user()
|
|
session.clear()
|
|
flash('You have been logged out.', 'info')
|
|
return redirect(url_for('auth.login')) |