""" control/panel/auth.py --------------------- Superadmin authentication for the control panel (MT-4). Uses Flask session directly (no Flask-Login) — the panel is a standalone app with no dependency on the main app's login_manager. Session key: 'sa_id' (superadmin.id when logged in) """ import logging from flask import ( Blueprint, render_template, redirect, url_for, request, session, flash, ) from control.base import control_session from control.models import Superadmin, TenantAudit from control.time_utils import now_eastern from control import mfa from control.panel.decorators import superadmin_required logger = logging.getLogger(__name__) bp = Blueprint('auth', __name__) def _log_audit(action, superadmin_id=None, tenant_id=None, details=None): """Write a TenantAudit row for superadmin actions.""" try: ip = (request.headers.get('X-Forwarded-For', '').split(',')[0].strip() or request.remote_addr) with control_session() as s: s.add(TenantAudit( superadmin_id=superadmin_id, action=action[:50], tenant_id=tenant_id, details=details, ip_address=(ip or '')[:45], created_at=now_eastern(), )) except Exception as exc: logger.error('TenantAudit write failed: %s', exc) # ── Login ───────────────────────────────────────────────────────────────────── @bp.route('/login', methods=['GET', 'POST']) def login(): if session.get('sa_id'): return redirect(url_for('tenants.list_tenants')) error = None if request.method == 'POST': username = (request.form.get('username') or '').strip() password = request.form.get('password') or '' with control_session() as s: sa = s.query(Superadmin).filter_by(username=username, active=True).first() if sa and sa.check_password(password): # ── Two-factor gate (control0005) ─────────────────────────── # Defer session establishment to the second-factor step when # this superadmin has TOTP enabled. if sa.mfa_enabled and sa.mfa_secret: session['sa_mfa_pending_id'] = sa.id session['sa_mfa_pending_username'] = sa.username session['sa_mfa_next'] = _safe_next(request.args.get('next')) logger.info('PANEL | mfa_challenge | superadmin=%s', username) return redirect(url_for('auth.mfa_challenge')) session.permanent = True session['sa_id'] = sa.id session['sa_username'] = sa.username logger.info('PANEL | login | superadmin=%s', username) _log_audit('LOGIN', superadmin_id=sa.id, details=f'superadmin={username}') flash(f'Welcome, {sa.username}!', 'success') next_url = _safe_next(request.args.get('next')) return redirect(next_url) else: logger.warning('PANEL | login_fail | username=%s', username) error = 'Invalid credentials.' return render_template('panel/login.html', error=error) def _safe_next(target): """Only allow same-app relative redirects (open-redirect guard).""" if target and target.startswith('/') and not target.startswith('//'): return target return url_for('tenants.list_tenants') # ── Two-factor challenge (control0005) ────────────────────────────────────── @bp.route('/mfa', methods=['GET', 'POST']) def mfa_challenge(): """Second-factor step during panel login. Reached only after a correct password for an MFA-enabled superadmin (identity held pending in session).""" sa_id = session.get('sa_mfa_pending_id') if not sa_id: return redirect(url_for('auth.login')) error = None if request.method == 'POST': code = request.form.get('code', '') use_recovery = bool(request.form.get('recovery')) verified, via, uname = False, 'totp', None with control_session() as s: sa = s.get(Superadmin, sa_id) if sa is None or not sa.active or not sa.mfa_enabled: session.pop('sa_mfa_pending_id', None) return redirect(url_for('auth.login')) if use_recovery: matched, remaining = mfa.check_and_consume_recovery( sa.mfa_recovery_codes, code) if matched: sa.mfa_recovery_codes = remaining # committed on block exit verified, via = True, 'recovery' elif mfa.verify_totp(sa.mfa_secret, code): verified = True uname = sa.username if verified: session.pop('sa_mfa_pending_id', None) session.pop('sa_mfa_pending_username', None) next_url = session.pop('sa_mfa_next', None) or url_for('tenants.list_tenants') session.permanent = True session['sa_id'] = sa_id session['sa_username'] = uname logger.info('PANEL | login | superadmin=%s | 2fa=%s', uname, via) _log_audit('LOGIN', superadmin_id=sa_id, details=f'superadmin={uname}; 2fa={via}') flash(f'Welcome, {uname}!', 'success') return redirect(next_url) logger.warning('PANEL | mfa_fail | superadmin_id=%s | recovery=%s', sa_id, use_recovery) error = 'Invalid verification code.' return render_template('panel/mfa_challenge.html', error=error) # ── Logout ──────────────────────────────────────────────────────────────────── @bp.route('/logout') def logout(): sa_id = session.get('sa_id') sa_name = session.get('sa_username', '?') session.clear() logger.info('PANEL | logout | superadmin=%s', sa_name) if sa_id: _log_audit('LOGOUT', superadmin_id=sa_id, details=f'superadmin={sa_name}') flash('Logged out.', 'info') return redirect(url_for('auth.login')) # ── Two-factor enrollment / management (control0005) ──────────────────────── @bp.route('/security') @superadmin_required def security(): """Superadmin security page — two-factor status + enable/disable controls.""" sa_id = session.get('sa_id') with control_session() as s: sa = s.get(Superadmin, sa_id) mfa_enabled = bool(sa and sa.mfa_enabled) recovery_remaining = len(sa.mfa_recovery_codes or []) if sa else 0 return render_template('panel/security.html', mfa_enabled=mfa_enabled, recovery_remaining=recovery_remaining, sa_username=session.get('sa_username')) @bp.route('/mfa/setup', methods=['GET', 'POST']) @superadmin_required def mfa_setup(): """Enroll the logged-in superadmin in TOTP. Candidate secret is held in the session until a valid code proves enrollment, so a half-finished setup can never lock the operator out.""" sa_id = session.get('sa_id') with control_session() as s: sa = s.get(Superadmin, sa_id) already = bool(sa and sa.mfa_enabled) account_name = sa.email or sa.username if sa else 'superadmin' if already: flash('Two-factor is already enabled on your account.', 'info') return redirect(url_for('auth.security')) if request.method == 'POST': secret = session.get('sa_setup_secret') code = request.form.get('code', '') if secret and mfa.verify_totp(secret, code): plaintext, hashed = mfa.generate_recovery_codes() with control_session() as s: sa = s.get(Superadmin, sa_id) sa.mfa_secret = secret sa.mfa_enabled = True sa.mfa_recovery_codes = hashed session.pop('sa_setup_secret', None) logger.info('PANEL | mfa_enabled | superadmin_id=%s', sa_id) _log_audit('MFA_ENABLE', superadmin_id=sa_id, details='enabled 2FA') return render_template('panel/mfa_recovery.html', codes=plaintext, sa_username=session.get('sa_username')) flash('That code did not match. Check your device clock and try again.', 'danger') secret = session.get('sa_setup_secret') or mfa.new_secret() session['sa_setup_secret'] = secret uri = mfa.provisioning_uri(secret, account_name) return render_template('panel/mfa_setup.html', secret=secret, qr_svg=mfa.qr_svg(uri), sa_username=session.get('sa_username')) @bp.route('/mfa/disable', methods=['POST']) @superadmin_required def mfa_disable(): """Turn off two-factor. Requires a current TOTP code or the password so a hijacked session cannot silently strip 2FA.""" sa_id = session.get('sa_id') code = request.form.get('code', '') pw = request.form.get('password', '') with control_session() as s: sa = s.get(Superadmin, sa_id) if not sa or not sa.mfa_enabled: return redirect(url_for('auth.security')) if not (mfa.verify_totp(sa.mfa_secret, code) or (pw and sa.check_password(pw))): flash('Enter a valid authenticator code or your password to disable 2FA.', 'danger') return redirect(url_for('auth.security')) sa.mfa_enabled = False sa.mfa_secret = None sa.mfa_recovery_codes = None logger.info('PANEL | mfa_disabled | superadmin_id=%s', sa_id) _log_audit('MFA_DISABLE', superadmin_id=sa_id, details='disabled 2FA') flash('Two-factor authentication has been disabled.', 'success') return redirect(url_for('tenants.list_tenants'))