team-tryouts/app/routes/auth.py

"""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'))
        # 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)
        # flush, not commit: the id is needed for the gamertag rows below,
        # and signing up is one operation. Committing here made it two, so a
        # failure while writing the gamertags left an account whose declared
        # games were silently absent (ARCH-006).
        db.session.flush()

        # 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'))