GGThedandClaude Opus 5
7cec18c139
QUA-002, premiere moitie. **Ce commit ne fait que reformater** : aucun changement de comportement, aucune ligne de logique touchee. 72 fichiers, 4 restaient deja conformes. Il est isole exprès, pour que `git log -p` sur les commits voisins reste lisible. `quote-style = "preserve"` etait deja pose dans pyproject.toml, ce qui evite le brassage guillemets simples / doubles : le diff porte sur les retours a la ligne, l indentation des appels longs et les virgules finales, pas sur le style de chaine. Verification : 263 tests passent avant et apres, ruff check propre. L activation en CI arrive dans le commit suivant, separement, pour que ce diff-ci ne contienne rien d autre. Co-Authored-By: Claude Opus 5 <[email protected]>
601 lines
22 KiB
Python
601 lines
22 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
|
|
import secrets
|
|
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 app.i18n import LOCALE_SESSION_KEY
|
|
from app.logging_config import log_auth_event
|
|
from flask_babel import gettext as _
|
|
from marshmallow import ValidationError
|
|
from urllib.parse import urlparse, urlencode
|
|
import requests
|
|
|
|
#: Session key holding the pending OAuth2 anti-forgery token.
|
|
DISCORD_STATE_KEY = 'discord_oauth_state'
|
|
|
|
# Failed-attempt tracking. The tally is kept for the audit trail and for the
|
|
# cool-off marker below; it no longer refuses a correct password (SEC-018).
|
|
MAX_LOGIN_ATTEMPTS = 5
|
|
LOCKOUT_DURATION_MINUTES = 15
|
|
#: Ceiling on the doubling cool-off window.
|
|
MAX_LOCKOUT_MINUTES = 240
|
|
|
|
#: Hash of a value nobody can submit. Verifying against it when the username
|
|
#: is unknown makes that path cost the same scrypt work as a real one, so the
|
|
#: response time stops telling a caller which usernames exist (SEC-017).
|
|
_ABSENT_USER_HASH = None
|
|
|
|
# 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).
|
|
|
|
Accepts an absolute URL on this host, or a path beginning with exactly
|
|
one slash. Everything else is refused, including the two forms that
|
|
read differently to urlparse and to a browser:
|
|
|
|
/\\evil.com several browsers normalise the backslash to a slash,
|
|
turning this into the protocol-relative //evil.com.
|
|
urlparse reports no netloc at all, so the old check
|
|
let it through and the redirect left the site.
|
|
/\\n//evil.com control characters are stripped before parsing.
|
|
|
|
Args:
|
|
url: The URL to validate.
|
|
|
|
Returns:
|
|
bool: True if the URL is safe.
|
|
"""
|
|
if not url:
|
|
return False
|
|
if any(ord(char) < 0x20 or char in '\\\x7f' for char in url):
|
|
return False
|
|
|
|
parsed = urlparse(url)
|
|
if parsed.netloc:
|
|
return parsed.netloc == request.host and parsed.scheme in ('', 'http', 'https')
|
|
# Relative targets must be rooted. 'dashboard' or 'javascript:...' are
|
|
# not paths on this site.
|
|
return url.startswith('/')
|
|
|
|
|
|
def _absent_user_hash():
|
|
"""A hash to verify against when the submitted username does not exist.
|
|
|
|
check_password() used to be reached only when a user row was found, so
|
|
an unknown username answered as fast as the database lookup, and a known
|
|
one as slowly as scrypt. The gap is measurable and enumerates accounts.
|
|
Computed once per process, from a random secret, so no submitted password
|
|
can ever match it.
|
|
"""
|
|
global _ABSENT_USER_HASH
|
|
if _ABSENT_USER_HASH is None:
|
|
_ABSENT_USER_HASH = hash_password(secrets.token_urlsafe(32))
|
|
return _ABSENT_USER_HASH
|
|
|
|
|
|
def cooloff_minutes(failed_attempts):
|
|
"""Length of the cool-off window earned by this many failed attempts.
|
|
|
|
Doubles every MAX_LOGIN_ATTEMPTS further failures, up to a ceiling.
|
|
|
|
Args:
|
|
failed_attempts: Consecutive failures recorded on the account.
|
|
|
|
Returns:
|
|
int: Minutes.
|
|
"""
|
|
steps = max(failed_attempts // MAX_LOGIN_ATTEMPTS - 1, 0)
|
|
return min(LOCKOUT_DURATION_MINUTES * (2**steps), MAX_LOCKOUT_MINUTES)
|
|
|
|
|
|
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.
|
|
|
|
GET: Render the login form.
|
|
POST: Authenticate user credentials, with audit logging.
|
|
|
|
Failed attempts are counted and open a cool-off window, recorded in
|
|
``locked_until`` and in the authentication log. The window does not
|
|
refuse correct credentials: when it did, five wrong guesses against a
|
|
known username took that account out of service for fifteen minutes,
|
|
repeatably, and on a president's account that meant no administration
|
|
at all. Guess rate is bounded by the rate limit on this view.
|
|
|
|
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(_('%(field)s: %(msg)s', field=field, msg=msg), 'danger')
|
|
return render_template('pages/login.html')
|
|
|
|
username = validated['username']
|
|
password = validated['password']
|
|
user = User.query.filter_by(username=username).first()
|
|
|
|
# Verified before anything else is decided, and on both branches.
|
|
# Reaching this only when a row exists made the response time a
|
|
# reliable oracle for which usernames are registered (SEC-017).
|
|
credentials_ok = check_password(
|
|
user.password_hash if user else _absent_user_hash(), password
|
|
)
|
|
|
|
if user and credentials_ok:
|
|
if not user.is_active_account:
|
|
log_auth_event('login.rejected.deactivated', username=username, user_id=user.id)
|
|
flash(_('This account has been deactivated.'), 'danger')
|
|
return render_template('pages/login.html')
|
|
|
|
# Correct credentials clear the tally, cool-off window included.
|
|
# The window used to refuse them too, which is what turned it
|
|
# into a way to lock a known account out at will (SEC-018).
|
|
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).
|
|
#
|
|
# The language choice is carried across too. Someone who reads the
|
|
# login page in English and signs in would otherwise be dropped
|
|
# back into French — the preference lives in the session, and
|
|
# clearing it discards a decision the user just made.
|
|
_preserved = {
|
|
key: session[key] for key in ('csrf_token', LOCALE_SESSION_KEY) if key in session
|
|
}
|
|
session.clear()
|
|
session.update(_preserved)
|
|
|
|
# Mark the session permanent so PERMANENT_SESSION_LIFETIME applies.
|
|
# Without this, Flask emits a browser-session cookie with no expiry
|
|
# and the configured lifetime is silently ignored.
|
|
session.permanent = True
|
|
|
|
login_user(user)
|
|
log_auth_event('login.success', username=user.username, user_id=user.id, role=user.role)
|
|
|
|
# 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(_('Welcome back, %(username)s!', username=user.username), 'success')
|
|
return redirect(next_page) if next_page else redirect(url_for('main.dashboard'))
|
|
else:
|
|
# One message for every failure. The old code said "N attempts
|
|
# remaining" to a real account and "check username and password"
|
|
# to an unknown one, which listed the club's accounts to anyone
|
|
# who asked (SEC-017).
|
|
if user:
|
|
user.failed_login_attempts += 1
|
|
log_auth_event(
|
|
'login.failure',
|
|
username=username,
|
|
user_id=user.id,
|
|
attempts=user.failed_login_attempts,
|
|
)
|
|
if user.failed_login_attempts >= MAX_LOGIN_ATTEMPTS:
|
|
minutes = cooloff_minutes(user.failed_login_attempts)
|
|
user.locked_until = datetime.utcnow() + timedelta(minutes=minutes)
|
|
log_auth_event(
|
|
'account.throttled',
|
|
username=username,
|
|
user_id=user.id,
|
|
minutes=minutes,
|
|
attempts=user.failed_login_attempts,
|
|
)
|
|
db.session.commit()
|
|
else:
|
|
log_auth_event('login.failure.unknown_user', username=username)
|
|
|
|
flash(
|
|
_(
|
|
'Login unsuccessful. Please check your username and '
|
|
'password, or ask a president for help.'
|
|
),
|
|
'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(_('%(field)s: %(msg)s', field=field, msg=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)
|
|
|
|
log_auth_event('account.registered', username=user.username, user_id=user.id)
|
|
|
|
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.
|
|
"""
|
|
# DISCORD_REDIRECT_URI is checked too: quoting it when unset used to
|
|
# raise inside the query builder rather than report a configuration error.
|
|
if not DISCORD_CLIENT_ID or not DISCORD_REDIRECT_URI:
|
|
flash(_('Discord OAuth2 is not configured.'), 'danger')
|
|
return redirect(url_for('auth.register'))
|
|
|
|
# Anti-forgery token, required by RFC 6749 §10.12. Without it, an
|
|
# attacker could have the victim's browser consume an authorization code
|
|
# obtained for the attacker's own Discord account, silently binding that
|
|
# identity to the victim's registration form.
|
|
state = secrets.token_urlsafe(32)
|
|
session[DISCORD_STATE_KEY] = state
|
|
|
|
params = {
|
|
'client_id': DISCORD_CLIENT_ID,
|
|
'redirect_uri': DISCORD_REDIRECT_URI,
|
|
'response_type': 'code',
|
|
'scope': 'identify connections',
|
|
'state': state,
|
|
}
|
|
auth_url = f'{DISCORD_API_BASE}/oauth2/authorize?{urlencode(params)}'
|
|
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.
|
|
"""
|
|
# The state is consumed whatever happens next: a token is single-use, and
|
|
# leaving it in the session would allow a replay.
|
|
expected_state = session.pop(DISCORD_STATE_KEY, None)
|
|
received_state = request.args.get('state', '')
|
|
|
|
if not expected_state or not secrets.compare_digest(expected_state, received_state):
|
|
flash(
|
|
_(
|
|
'Discord authorization could not be verified. '
|
|
'Please start the connection again from this page.'
|
|
),
|
|
'danger',
|
|
)
|
|
return redirect(url_for('auth.register'))
|
|
|
|
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:
|
|
flash(_('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', methods=['POST'])
|
|
@login_required
|
|
def logout():
|
|
"""Log out the current user and clear the session.
|
|
|
|
POST, not GET: a GET route is not covered by CSRF protection, so any
|
|
page on the internet could sign a user out with an <img> tag pointing
|
|
here. A nuisance rather than a compromise, but it costs one form to
|
|
close (SEC-019).
|
|
|
|
Clears the user session and regenerates session ID to prevent
|
|
session fixation/replay after logout.
|
|
|
|
Returns:
|
|
Response: Redirect to login page with logout message.
|
|
"""
|
|
log_auth_event('logout', username=current_user.username, user_id=current_user.id)
|
|
logout_user()
|
|
# Same reasoning as at login: the language is a display preference, not
|
|
# session state belonging to the account being signed out.
|
|
_locale = session.get(LOCALE_SESSION_KEY)
|
|
session.clear()
|
|
if _locale:
|
|
session[LOCALE_SESSION_KEY] = _locale
|
|
flash(_('You have been logged out.'), 'info')
|
|
return redirect(url_for('auth.login'))
|