244 lines
10 KiB
Python
244 lines
10 KiB
Python
"""
|
|
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'))
|