Refactoring - Part 1
1. Konfiguration extrahiert (config.py)
- Alle App-Einstellungen zentralisiert
- Flask-Konfiguration, Datenbank, Backup, Rate-Limiting
- 576 Zeilen Code reduziert
2. Datenbank-Layer (db.py)
- Connection Management mit Context Managers
- Helper-Funktionen für Queries
- Saubere Fehlerbehandlung
3. Auth-Module (auth/)
- decorators.py - Login-Required mit Session-Timeout
- password.py - Bcrypt Hashing
- two_factor.py - TOTP, QR-Codes, Backup-Codes
- rate_limiting.py - IP-Blocking, Login-Versuche
4. Utility-Module (utils/)
- audit.py - Audit-Logging
- backup.py - Verschlüsselte Backups
- license.py - Lizenzschlüssel-Generierung
- export.py - Excel-Export
- network.py - IP-Ermittlung
- recaptcha.py - reCAPTCHA-Verifikation
5. Models (models.py)
- User-Model-Funktionen
Dieser Commit ist enthalten in:
2
v2_adminpanel/routes/__init__.py
Normale Datei
2
v2_adminpanel/routes/__init__.py
Normale Datei
@@ -0,0 +1,2 @@
|
||||
# Routes module initialization
|
||||
# This module contains all Flask blueprints organized by functionality
|
||||
540
v2_adminpanel/routes/admin_routes.py
Normale Datei
540
v2_adminpanel/routes/admin_routes.py
Normale Datei
@@ -0,0 +1,540 @@
|
||||
import os
|
||||
from datetime import datetime, timedelta
|
||||
from zoneinfo import ZoneInfo
|
||||
from pathlib import Path
|
||||
from flask import Blueprint, render_template, request, redirect, session, url_for, flash, send_file, jsonify
|
||||
|
||||
import config
|
||||
from auth.decorators import login_required
|
||||
from utils.audit import log_audit
|
||||
from utils.backup import create_backup, restore_backup
|
||||
from utils.network import get_client_ip
|
||||
from db import get_connection, get_db_connection, get_db_cursor, execute_query
|
||||
from utils.export import create_excel_export, prepare_audit_export_data
|
||||
|
||||
# Create Blueprint
|
||||
admin_bp = Blueprint('admin', __name__)
|
||||
|
||||
|
||||
@admin_bp.route("/")
|
||||
@login_required
|
||||
def dashboard():
|
||||
conn = get_connection()
|
||||
cur = conn.cursor()
|
||||
|
||||
try:
|
||||
# Hole Statistiken
|
||||
# Anzahl aktiver Lizenzen
|
||||
cur.execute("SELECT COUNT(*) FROM licenses WHERE active = true")
|
||||
active_licenses = cur.fetchone()[0]
|
||||
|
||||
# Anzahl Kunden
|
||||
cur.execute("SELECT COUNT(*) FROM customers")
|
||||
total_customers = cur.fetchone()[0]
|
||||
|
||||
# Anzahl aktiver Sessions
|
||||
cur.execute("SELECT COUNT(*) FROM sessions WHERE active = true")
|
||||
active_sessions = cur.fetchone()[0]
|
||||
|
||||
# Top 10 Lizenzen nach Nutzung (letzte 30 Tage)
|
||||
cur.execute("""
|
||||
SELECT
|
||||
l.license_key,
|
||||
c.name as customer_name,
|
||||
COUNT(DISTINCT s.id) as session_count,
|
||||
COUNT(DISTINCT s.username) as unique_users,
|
||||
MAX(s.last_activity) as last_activity
|
||||
FROM licenses l
|
||||
LEFT JOIN customers c ON l.customer_id = c.id
|
||||
LEFT JOIN sessions s ON l.license_key = s.license_key
|
||||
AND s.login_time >= CURRENT_TIMESTAMP - INTERVAL '30 days'
|
||||
GROUP BY l.license_key, c.name
|
||||
ORDER BY session_count DESC
|
||||
LIMIT 10
|
||||
""")
|
||||
top_licenses = cur.fetchall()
|
||||
|
||||
# Letzte 10 Aktivitäten aus dem Audit Log
|
||||
cur.execute("""
|
||||
SELECT
|
||||
id,
|
||||
timestamp AT TIME ZONE 'Europe/Berlin' as timestamp,
|
||||
username,
|
||||
action,
|
||||
entity_type,
|
||||
entity_id,
|
||||
additional_info
|
||||
FROM audit_log
|
||||
ORDER BY timestamp DESC
|
||||
LIMIT 10
|
||||
""")
|
||||
recent_activities = cur.fetchall()
|
||||
|
||||
# Lizenztyp-Verteilung
|
||||
cur.execute("""
|
||||
SELECT
|
||||
CASE
|
||||
WHEN is_test_license THEN 'Test'
|
||||
ELSE 'Full'
|
||||
END as license_type,
|
||||
COUNT(*) as count
|
||||
FROM licenses
|
||||
GROUP BY is_test_license
|
||||
""")
|
||||
license_distribution = cur.fetchall()
|
||||
|
||||
# Sessions nach Stunden (letzte 24h)
|
||||
cur.execute("""
|
||||
WITH hours AS (
|
||||
SELECT generate_series(
|
||||
CURRENT_TIMESTAMP - INTERVAL '23 hours',
|
||||
CURRENT_TIMESTAMP,
|
||||
INTERVAL '1 hour'
|
||||
) AS hour
|
||||
)
|
||||
SELECT
|
||||
TO_CHAR(hours.hour AT TIME ZONE 'Europe/Berlin', 'HH24:00') as hour_label,
|
||||
COUNT(DISTINCT s.id) as session_count
|
||||
FROM hours
|
||||
LEFT JOIN sessions s ON
|
||||
s.login_time >= hours.hour AND
|
||||
s.login_time < hours.hour + INTERVAL '1 hour'
|
||||
GROUP BY hours.hour
|
||||
ORDER BY hours.hour
|
||||
""")
|
||||
hourly_sessions = cur.fetchall()
|
||||
|
||||
# System-Status
|
||||
cur.execute("SELECT pg_database_size(current_database())")
|
||||
db_size = cur.fetchone()[0]
|
||||
|
||||
# Letzte Backup-Info
|
||||
cur.execute("""
|
||||
SELECT filename, created_at, filesize, status
|
||||
FROM backup_history
|
||||
WHERE status = 'success'
|
||||
ORDER BY created_at DESC
|
||||
LIMIT 1
|
||||
""")
|
||||
last_backup = cur.fetchone()
|
||||
|
||||
# Resource Statistiken
|
||||
cur.execute("""
|
||||
SELECT
|
||||
COUNT(*) FILTER (WHERE status = 'available') as available,
|
||||
COUNT(*) FILTER (WHERE status = 'in_use') as in_use,
|
||||
COUNT(*) FILTER (WHERE status = 'quarantine') as quarantine,
|
||||
COUNT(*) as total
|
||||
FROM resources
|
||||
""")
|
||||
resource_stats = cur.fetchone()
|
||||
|
||||
return render_template('dashboard.html',
|
||||
active_licenses=active_licenses,
|
||||
total_customers=total_customers,
|
||||
active_sessions=active_sessions,
|
||||
top_licenses=top_licenses,
|
||||
recent_activities=recent_activities,
|
||||
license_distribution=license_distribution,
|
||||
hourly_sessions=hourly_sessions,
|
||||
db_size=db_size,
|
||||
last_backup=last_backup,
|
||||
resource_stats=resource_stats,
|
||||
username=session.get('username'))
|
||||
|
||||
finally:
|
||||
cur.close()
|
||||
conn.close()
|
||||
|
||||
|
||||
@admin_bp.route("/audit")
|
||||
@login_required
|
||||
def audit_log():
|
||||
page = request.args.get('page', 1, type=int)
|
||||
per_page = 50
|
||||
search = request.args.get('search', '')
|
||||
action_filter = request.args.get('action', '')
|
||||
entity_filter = request.args.get('entity', '')
|
||||
|
||||
conn = get_connection()
|
||||
cur = conn.cursor()
|
||||
|
||||
try:
|
||||
# Base query
|
||||
query = """
|
||||
SELECT
|
||||
id,
|
||||
timestamp AT TIME ZONE 'Europe/Berlin' as timestamp,
|
||||
username,
|
||||
action,
|
||||
entity_type,
|
||||
entity_id,
|
||||
old_values::text,
|
||||
new_values::text,
|
||||
ip_address,
|
||||
user_agent,
|
||||
additional_info
|
||||
FROM audit_log
|
||||
WHERE 1=1
|
||||
"""
|
||||
params = []
|
||||
|
||||
# Suchfilter
|
||||
if search:
|
||||
query += """ AND (
|
||||
username ILIKE %s OR
|
||||
action ILIKE %s OR
|
||||
entity_type ILIKE %s OR
|
||||
additional_info ILIKE %s OR
|
||||
ip_address ILIKE %s
|
||||
)"""
|
||||
search_param = f"%{search}%"
|
||||
params.extend([search_param] * 5)
|
||||
|
||||
# Action Filter
|
||||
if action_filter:
|
||||
query += " AND action = %s"
|
||||
params.append(action_filter)
|
||||
|
||||
# Entity Filter
|
||||
if entity_filter:
|
||||
query += " AND entity_type = %s"
|
||||
params.append(entity_filter)
|
||||
|
||||
# Count total
|
||||
count_query = f"SELECT COUNT(*) FROM ({query}) as filtered"
|
||||
cur.execute(count_query, params)
|
||||
total_count = cur.fetchone()[0]
|
||||
|
||||
# Add pagination
|
||||
query += " ORDER BY timestamp DESC LIMIT %s OFFSET %s"
|
||||
params.extend([per_page, (page - 1) * per_page])
|
||||
|
||||
cur.execute(query, params)
|
||||
logs = cur.fetchall()
|
||||
|
||||
# Get unique actions and entities for filters
|
||||
cur.execute("SELECT DISTINCT action FROM audit_log ORDER BY action")
|
||||
actions = [row[0] for row in cur.fetchall()]
|
||||
|
||||
cur.execute("SELECT DISTINCT entity_type FROM audit_log ORDER BY entity_type")
|
||||
entities = [row[0] for row in cur.fetchall()]
|
||||
|
||||
# Pagination info
|
||||
total_pages = (total_count + per_page - 1) // per_page
|
||||
|
||||
# Convert to dictionaries for easier template access
|
||||
audit_logs = []
|
||||
for log in logs:
|
||||
audit_logs.append({
|
||||
'id': log[0],
|
||||
'timestamp': log[1],
|
||||
'username': log[2],
|
||||
'action': log[3],
|
||||
'entity_type': log[4],
|
||||
'entity_id': log[5],
|
||||
'old_values': log[6],
|
||||
'new_values': log[7],
|
||||
'ip_address': log[8],
|
||||
'user_agent': log[9],
|
||||
'additional_info': log[10]
|
||||
})
|
||||
|
||||
return render_template('audit_log.html',
|
||||
logs=audit_logs,
|
||||
page=page,
|
||||
total_pages=total_pages,
|
||||
total_count=total_count,
|
||||
search=search,
|
||||
action_filter=action_filter,
|
||||
entity_filter=entity_filter,
|
||||
actions=actions,
|
||||
entities=entities,
|
||||
username=session.get('username'))
|
||||
|
||||
finally:
|
||||
cur.close()
|
||||
conn.close()
|
||||
|
||||
|
||||
@admin_bp.route("/backups")
|
||||
@login_required
|
||||
def backups():
|
||||
conn = get_connection()
|
||||
cur = conn.cursor()
|
||||
|
||||
try:
|
||||
# Hole alle Backups
|
||||
cur.execute("""
|
||||
SELECT
|
||||
id,
|
||||
filename,
|
||||
created_at AT TIME ZONE 'Europe/Berlin' as created_at,
|
||||
filesize,
|
||||
backup_type,
|
||||
status,
|
||||
created_by,
|
||||
duration_seconds,
|
||||
tables_count,
|
||||
records_count,
|
||||
error_message,
|
||||
is_encrypted
|
||||
FROM backup_history
|
||||
ORDER BY created_at DESC
|
||||
""")
|
||||
backups = cur.fetchall()
|
||||
|
||||
# Prüfe ob Dateien noch existieren
|
||||
backups_with_status = []
|
||||
for backup in backups:
|
||||
backup_dict = {
|
||||
'id': backup[0],
|
||||
'filename': backup[1],
|
||||
'created_at': backup[2],
|
||||
'filesize': backup[3],
|
||||
'backup_type': backup[4],
|
||||
'status': backup[5],
|
||||
'created_by': backup[6],
|
||||
'duration_seconds': backup[7],
|
||||
'tables_count': backup[8],
|
||||
'records_count': backup[9],
|
||||
'error_message': backup[10],
|
||||
'is_encrypted': backup[11],
|
||||
'file_exists': False
|
||||
}
|
||||
|
||||
# Prüfe ob Datei existiert
|
||||
if backup[1]: # filename
|
||||
filepath = config.BACKUP_DIR / backup[1]
|
||||
backup_dict['file_exists'] = filepath.exists()
|
||||
|
||||
backups_with_status.append(backup_dict)
|
||||
|
||||
return render_template('backups.html',
|
||||
backups=backups_with_status,
|
||||
username=session.get('username'))
|
||||
|
||||
finally:
|
||||
cur.close()
|
||||
conn.close()
|
||||
|
||||
|
||||
@admin_bp.route("/backup/create", methods=["POST"])
|
||||
@login_required
|
||||
def create_backup_route():
|
||||
"""Manuelles Backup erstellen"""
|
||||
success, result = create_backup(backup_type="manual", created_by=session.get('username'))
|
||||
|
||||
if success:
|
||||
flash(f'Backup erfolgreich erstellt: {result}', 'success')
|
||||
else:
|
||||
flash(f'Backup fehlgeschlagen: {result}', 'error')
|
||||
|
||||
return redirect(url_for('admin.backups'))
|
||||
|
||||
|
||||
@admin_bp.route("/backup/restore/<int:backup_id>", methods=["POST"])
|
||||
@login_required
|
||||
def restore_backup_route(backup_id):
|
||||
"""Backup wiederherstellen"""
|
||||
encryption_key = request.form.get('encryption_key')
|
||||
|
||||
success, message = restore_backup(backup_id, encryption_key)
|
||||
|
||||
if success:
|
||||
flash(message, 'success')
|
||||
else:
|
||||
flash(f'Wiederherstellung fehlgeschlagen: {message}', 'error')
|
||||
|
||||
return redirect(url_for('admin.backups'))
|
||||
|
||||
|
||||
@admin_bp.route("/backup/download/<int:backup_id>")
|
||||
@login_required
|
||||
def download_backup(backup_id):
|
||||
"""Backup herunterladen"""
|
||||
conn = get_connection()
|
||||
cur = conn.cursor()
|
||||
|
||||
try:
|
||||
# Hole Backup-Info
|
||||
cur.execute("SELECT filename, filepath FROM backup_history WHERE id = %s", (backup_id,))
|
||||
result = cur.fetchone()
|
||||
|
||||
if not result:
|
||||
flash('Backup nicht gefunden', 'error')
|
||||
return redirect(url_for('admin.backups'))
|
||||
|
||||
filename, filepath = result
|
||||
filepath = Path(filepath)
|
||||
|
||||
if not filepath.exists():
|
||||
flash('Backup-Datei nicht gefunden', 'error')
|
||||
return redirect(url_for('admin.backups'))
|
||||
|
||||
# Audit-Log
|
||||
log_audit('BACKUP_DOWNLOAD', 'backup', backup_id,
|
||||
additional_info=f"Backup heruntergeladen: {filename}")
|
||||
|
||||
return send_file(filepath, as_attachment=True, download_name=filename)
|
||||
|
||||
finally:
|
||||
cur.close()
|
||||
conn.close()
|
||||
|
||||
|
||||
@admin_bp.route("/backup/delete/<int:backup_id>", methods=["DELETE"])
|
||||
@login_required
|
||||
def delete_backup(backup_id):
|
||||
"""Backup löschen"""
|
||||
conn = get_connection()
|
||||
cur = conn.cursor()
|
||||
|
||||
try:
|
||||
# Hole Backup-Info
|
||||
cur.execute("SELECT filename, filepath FROM backup_history WHERE id = %s", (backup_id,))
|
||||
result = cur.fetchone()
|
||||
|
||||
if not result:
|
||||
return jsonify({'success': False, 'message': 'Backup nicht gefunden'}), 404
|
||||
|
||||
filename, filepath = result
|
||||
filepath = Path(filepath)
|
||||
|
||||
# Lösche Datei wenn vorhanden
|
||||
if filepath.exists():
|
||||
try:
|
||||
filepath.unlink()
|
||||
except Exception as e:
|
||||
return jsonify({'success': False, 'message': f'Fehler beim Löschen der Datei: {str(e)}'}), 500
|
||||
|
||||
# Lösche Datenbank-Eintrag
|
||||
cur.execute("DELETE FROM backup_history WHERE id = %s", (backup_id,))
|
||||
conn.commit()
|
||||
|
||||
# Audit-Log
|
||||
log_audit('BACKUP_DELETE', 'backup', backup_id,
|
||||
additional_info=f"Backup gelöscht: {filename}")
|
||||
|
||||
return jsonify({'success': True, 'message': 'Backup erfolgreich gelöscht'})
|
||||
|
||||
except Exception as e:
|
||||
conn.rollback()
|
||||
return jsonify({'success': False, 'message': str(e)}), 500
|
||||
|
||||
finally:
|
||||
cur.close()
|
||||
conn.close()
|
||||
|
||||
|
||||
@admin_bp.route("/security/blocked-ips")
|
||||
@login_required
|
||||
def blocked_ips():
|
||||
"""Zeigt gesperrte IP-Adressen"""
|
||||
conn = get_connection()
|
||||
cur = conn.cursor()
|
||||
|
||||
try:
|
||||
cur.execute("""
|
||||
SELECT
|
||||
ip_address,
|
||||
attempt_count,
|
||||
last_attempt AT TIME ZONE 'Europe/Berlin' as last_attempt,
|
||||
blocked_until AT TIME ZONE 'Europe/Berlin' as blocked_until,
|
||||
last_username_tried,
|
||||
last_error_message
|
||||
FROM login_attempts
|
||||
WHERE blocked_until IS NOT NULL AND blocked_until > CURRENT_TIMESTAMP
|
||||
ORDER BY blocked_until DESC
|
||||
""")
|
||||
blocked = cur.fetchall()
|
||||
|
||||
# Alle Login-Versuche (auch nicht gesperrte)
|
||||
cur.execute("""
|
||||
SELECT
|
||||
ip_address,
|
||||
attempt_count,
|
||||
last_attempt AT TIME ZONE 'Europe/Berlin' as last_attempt,
|
||||
blocked_until AT TIME ZONE 'Europe/Berlin' as blocked_until,
|
||||
last_username_tried,
|
||||
last_error_message
|
||||
FROM login_attempts
|
||||
ORDER BY last_attempt DESC
|
||||
LIMIT 100
|
||||
""")
|
||||
all_attempts = cur.fetchall()
|
||||
|
||||
return render_template('blocked_ips.html',
|
||||
blocked_ips=blocked,
|
||||
all_attempts=all_attempts,
|
||||
username=session.get('username'))
|
||||
|
||||
finally:
|
||||
cur.close()
|
||||
conn.close()
|
||||
|
||||
|
||||
@admin_bp.route("/security/unblock-ip", methods=["POST"])
|
||||
@login_required
|
||||
def unblock_ip():
|
||||
"""Entsperrt eine IP-Adresse"""
|
||||
ip_address = request.form.get('ip_address')
|
||||
|
||||
if not ip_address:
|
||||
flash('Keine IP-Adresse angegeben', 'error')
|
||||
return redirect(url_for('admin.blocked_ips'))
|
||||
|
||||
conn = get_connection()
|
||||
cur = conn.cursor()
|
||||
|
||||
try:
|
||||
cur.execute("""
|
||||
UPDATE login_attempts
|
||||
SET blocked_until = NULL
|
||||
WHERE ip_address = %s
|
||||
""", (ip_address,))
|
||||
|
||||
if cur.rowcount > 0:
|
||||
conn.commit()
|
||||
flash(f'IP-Adresse {ip_address} wurde entsperrt', 'success')
|
||||
log_audit('UNBLOCK_IP', 'security',
|
||||
additional_info=f"IP-Adresse entsperrt: {ip_address}")
|
||||
else:
|
||||
flash(f'IP-Adresse {ip_address} nicht gefunden', 'warning')
|
||||
|
||||
except Exception as e:
|
||||
conn.rollback()
|
||||
flash(f'Fehler beim Entsperren: {str(e)}', 'error')
|
||||
|
||||
finally:
|
||||
cur.close()
|
||||
conn.close()
|
||||
|
||||
return redirect(url_for('admin.blocked_ips'))
|
||||
|
||||
|
||||
@admin_bp.route("/security/clear-attempts", methods=["POST"])
|
||||
@login_required
|
||||
def clear_attempts():
|
||||
"""Löscht alle Login-Versuche"""
|
||||
conn = get_connection()
|
||||
cur = conn.cursor()
|
||||
|
||||
try:
|
||||
cur.execute("DELETE FROM login_attempts")
|
||||
count = cur.rowcount
|
||||
conn.commit()
|
||||
|
||||
flash(f'{count} Login-Versuche wurden gelöscht', 'success')
|
||||
log_audit('CLEAR_LOGIN_ATTEMPTS', 'security',
|
||||
additional_info=f"{count} Login-Versuche gelöscht")
|
||||
|
||||
except Exception as e:
|
||||
conn.rollback()
|
||||
flash(f'Fehler beim Löschen: {str(e)}', 'error')
|
||||
|
||||
finally:
|
||||
cur.close()
|
||||
conn.close()
|
||||
|
||||
return redirect(url_for('admin.blocked_ips'))
|
||||
377
v2_adminpanel/routes/auth_routes.py
Normale Datei
377
v2_adminpanel/routes/auth_routes.py
Normale Datei
@@ -0,0 +1,377 @@
|
||||
import time
|
||||
import json
|
||||
from datetime import datetime
|
||||
from zoneinfo import ZoneInfo
|
||||
from flask import Blueprint, render_template, request, redirect, session, url_for, flash, jsonify
|
||||
|
||||
import config
|
||||
from auth.decorators import login_required
|
||||
from auth.password import hash_password, verify_password
|
||||
from auth.two_factor import (
|
||||
generate_totp_secret, generate_qr_code, verify_totp,
|
||||
generate_backup_codes, hash_backup_code, verify_backup_code
|
||||
)
|
||||
from auth.rate_limiting import (
|
||||
check_ip_blocked, record_failed_attempt,
|
||||
reset_login_attempts, get_login_attempts
|
||||
)
|
||||
from utils.network import get_client_ip
|
||||
from utils.audit import log_audit
|
||||
from models import get_user_by_username
|
||||
from db import get_db_connection, get_db_cursor
|
||||
from utils.recaptcha import verify_recaptcha
|
||||
|
||||
# Create Blueprint
|
||||
auth_bp = Blueprint('auth', __name__)
|
||||
|
||||
|
||||
@auth_bp.route("/login", methods=["GET", "POST"])
|
||||
def login():
|
||||
# Timing-Attack Schutz - Start Zeit merken
|
||||
start_time = time.time()
|
||||
|
||||
# IP-Adresse ermitteln
|
||||
ip_address = get_client_ip()
|
||||
|
||||
# Prüfen ob IP gesperrt ist
|
||||
is_blocked, blocked_until = check_ip_blocked(ip_address)
|
||||
if is_blocked:
|
||||
time_remaining = (blocked_until - datetime.now(ZoneInfo("Europe/Berlin")).replace(tzinfo=None)).total_seconds() / 3600
|
||||
error_msg = f"IP GESPERRT! Noch {time_remaining:.1f} Stunden warten."
|
||||
return render_template("login.html", error=error_msg, error_type="blocked")
|
||||
|
||||
# Anzahl bisheriger Versuche
|
||||
attempt_count = get_login_attempts(ip_address)
|
||||
|
||||
if request.method == "POST":
|
||||
username = request.form.get("username")
|
||||
password = request.form.get("password")
|
||||
captcha_response = request.form.get("g-recaptcha-response")
|
||||
|
||||
# CAPTCHA-Prüfung nur wenn Keys konfiguriert sind
|
||||
recaptcha_site_key = config.RECAPTCHA_SITE_KEY
|
||||
if attempt_count >= config.CAPTCHA_AFTER_ATTEMPTS and recaptcha_site_key:
|
||||
if not captcha_response:
|
||||
# Timing-Attack Schutz
|
||||
elapsed = time.time() - start_time
|
||||
if elapsed < 1.0:
|
||||
time.sleep(1.0 - elapsed)
|
||||
return render_template("login.html",
|
||||
error="CAPTCHA ERFORDERLICH!",
|
||||
show_captcha=True,
|
||||
error_type="captcha",
|
||||
attempts_left=max(0, config.MAX_LOGIN_ATTEMPTS - attempt_count),
|
||||
recaptcha_site_key=recaptcha_site_key)
|
||||
|
||||
# CAPTCHA validieren
|
||||
if not verify_recaptcha(captcha_response):
|
||||
# Timing-Attack Schutz
|
||||
elapsed = time.time() - start_time
|
||||
if elapsed < 1.0:
|
||||
time.sleep(1.0 - elapsed)
|
||||
return render_template("login.html",
|
||||
error="CAPTCHA UNGÜLTIG! Bitte erneut versuchen.",
|
||||
show_captcha=True,
|
||||
error_type="captcha",
|
||||
attempts_left=max(0, config.MAX_LOGIN_ATTEMPTS - attempt_count),
|
||||
recaptcha_site_key=recaptcha_site_key)
|
||||
|
||||
# Check user in database first, fallback to env vars
|
||||
user = get_user_by_username(username)
|
||||
login_success = False
|
||||
needs_2fa = False
|
||||
|
||||
if user:
|
||||
# Database user authentication
|
||||
if verify_password(password, user['password_hash']):
|
||||
login_success = True
|
||||
needs_2fa = user['totp_enabled']
|
||||
else:
|
||||
# Fallback to environment variables for backward compatibility
|
||||
if username in config.ADMIN_USERS and password == config.ADMIN_USERS[username]:
|
||||
login_success = True
|
||||
|
||||
# Timing-Attack Schutz - Mindestens 1 Sekunde warten
|
||||
elapsed = time.time() - start_time
|
||||
if elapsed < 1.0:
|
||||
time.sleep(1.0 - elapsed)
|
||||
|
||||
if login_success:
|
||||
# Erfolgreicher Login
|
||||
if needs_2fa:
|
||||
# Store temporary session for 2FA verification
|
||||
session['temp_username'] = username
|
||||
session['temp_user_id'] = user['id']
|
||||
session['awaiting_2fa'] = True
|
||||
return redirect(url_for('auth.verify_2fa'))
|
||||
else:
|
||||
# Complete login without 2FA
|
||||
session.permanent = True # Aktiviert das Timeout
|
||||
session['logged_in'] = True
|
||||
session['username'] = username
|
||||
session['user_id'] = user['id'] if user else None
|
||||
session['last_activity'] = datetime.now(ZoneInfo("Europe/Berlin")).replace(tzinfo=None).isoformat()
|
||||
reset_login_attempts(ip_address)
|
||||
log_audit('LOGIN_SUCCESS', 'user',
|
||||
additional_info=f"Erfolgreiche Anmeldung von IP: {ip_address}")
|
||||
return redirect(url_for('admin.dashboard'))
|
||||
else:
|
||||
# Fehlgeschlagener Login
|
||||
error_message = record_failed_attempt(ip_address, username)
|
||||
new_attempt_count = get_login_attempts(ip_address)
|
||||
|
||||
# Prüfen ob jetzt gesperrt
|
||||
is_now_blocked, _ = check_ip_blocked(ip_address)
|
||||
if is_now_blocked:
|
||||
log_audit('LOGIN_BLOCKED', 'security',
|
||||
additional_info=f"IP {ip_address} wurde nach {config.MAX_LOGIN_ATTEMPTS} Versuchen gesperrt")
|
||||
|
||||
return render_template("login.html",
|
||||
error=error_message,
|
||||
show_captcha=(new_attempt_count >= config.CAPTCHA_AFTER_ATTEMPTS and config.RECAPTCHA_SITE_KEY),
|
||||
error_type="failed",
|
||||
attempts_left=max(0, config.MAX_LOGIN_ATTEMPTS - new_attempt_count),
|
||||
recaptcha_site_key=config.RECAPTCHA_SITE_KEY)
|
||||
|
||||
# GET Request
|
||||
return render_template("login.html",
|
||||
show_captcha=(attempt_count >= config.CAPTCHA_AFTER_ATTEMPTS and config.RECAPTCHA_SITE_KEY),
|
||||
attempts_left=max(0, config.MAX_LOGIN_ATTEMPTS - attempt_count),
|
||||
recaptcha_site_key=config.RECAPTCHA_SITE_KEY)
|
||||
|
||||
|
||||
@auth_bp.route("/logout")
|
||||
def logout():
|
||||
username = session.get('username', 'unknown')
|
||||
log_audit('LOGOUT', 'user', additional_info=f"Abmeldung")
|
||||
session.pop('logged_in', None)
|
||||
session.pop('username', None)
|
||||
session.pop('user_id', None)
|
||||
session.pop('temp_username', None)
|
||||
session.pop('temp_user_id', None)
|
||||
session.pop('awaiting_2fa', None)
|
||||
return redirect(url_for('auth.login'))
|
||||
|
||||
|
||||
@auth_bp.route("/verify-2fa", methods=["GET", "POST"])
|
||||
def verify_2fa():
|
||||
if not session.get('awaiting_2fa'):
|
||||
return redirect(url_for('auth.login'))
|
||||
|
||||
if request.method == "POST":
|
||||
token = request.form.get('token', '').replace(' ', '')
|
||||
username = session.get('temp_username')
|
||||
user_id = session.get('temp_user_id')
|
||||
|
||||
if not username or not user_id:
|
||||
flash('Session expired. Please login again.', 'error')
|
||||
return redirect(url_for('auth.login'))
|
||||
|
||||
user = get_user_by_username(username)
|
||||
if not user:
|
||||
flash('User not found.', 'error')
|
||||
return redirect(url_for('auth.login'))
|
||||
|
||||
# Check if it's a backup code
|
||||
if len(token) == 8 and token.isupper():
|
||||
# Try backup code
|
||||
backup_codes = json.loads(user['backup_codes']) if user['backup_codes'] else []
|
||||
if verify_backup_code(token, backup_codes):
|
||||
# Remove used backup code
|
||||
code_hash = hash_backup_code(token)
|
||||
backup_codes.remove(code_hash)
|
||||
|
||||
with get_db_connection() as conn:
|
||||
with get_db_cursor(conn) as cur:
|
||||
cur.execute("UPDATE users SET backup_codes = %s WHERE id = %s",
|
||||
(json.dumps(backup_codes), user_id))
|
||||
|
||||
# Complete login
|
||||
session.permanent = True
|
||||
session['logged_in'] = True
|
||||
session['username'] = username
|
||||
session['user_id'] = user_id
|
||||
session['last_activity'] = datetime.now(ZoneInfo("Europe/Berlin")).replace(tzinfo=None).isoformat()
|
||||
session.pop('temp_username', None)
|
||||
session.pop('temp_user_id', None)
|
||||
session.pop('awaiting_2fa', None)
|
||||
|
||||
flash('Login successful using backup code. Please generate new backup codes.', 'warning')
|
||||
log_audit('LOGIN_2FA_BACKUP', 'user', additional_info=f"2FA login with backup code")
|
||||
return redirect(url_for('admin.dashboard'))
|
||||
else:
|
||||
# Try TOTP token
|
||||
if verify_totp(user['totp_secret'], token):
|
||||
# Complete login
|
||||
session.permanent = True
|
||||
session['logged_in'] = True
|
||||
session['username'] = username
|
||||
session['user_id'] = user_id
|
||||
session['last_activity'] = datetime.now(ZoneInfo("Europe/Berlin")).replace(tzinfo=None).isoformat()
|
||||
session.pop('temp_username', None)
|
||||
session.pop('temp_user_id', None)
|
||||
session.pop('awaiting_2fa', None)
|
||||
|
||||
log_audit('LOGIN_2FA_SUCCESS', 'user', additional_info=f"2FA login successful")
|
||||
return redirect(url_for('admin.dashboard'))
|
||||
|
||||
# Failed verification
|
||||
with get_db_connection() as conn:
|
||||
with get_db_cursor(conn) as cur:
|
||||
cur.execute("UPDATE users SET failed_2fa_attempts = failed_2fa_attempts + 1, last_failed_2fa = %s WHERE id = %s",
|
||||
(datetime.now(), user_id))
|
||||
|
||||
flash('Invalid authentication code. Please try again.', 'error')
|
||||
log_audit('LOGIN_2FA_FAILED', 'user', additional_info=f"Failed 2FA attempt")
|
||||
|
||||
return render_template('verify_2fa.html')
|
||||
|
||||
|
||||
@auth_bp.route("/profile")
|
||||
@login_required
|
||||
def profile():
|
||||
user = get_user_by_username(session['username'])
|
||||
if not user:
|
||||
# For environment-based users, redirect with message
|
||||
flash('Bitte führen Sie das Migrations-Script aus, um Passwort-Änderung und 2FA zu aktivieren.', 'info')
|
||||
return redirect(url_for('admin.dashboard'))
|
||||
return render_template('profile.html', user=user)
|
||||
|
||||
|
||||
@auth_bp.route("/profile/change-password", methods=["POST"])
|
||||
@login_required
|
||||
def change_password():
|
||||
current_password = request.form.get('current_password')
|
||||
new_password = request.form.get('new_password')
|
||||
confirm_password = request.form.get('confirm_password')
|
||||
|
||||
user = get_user_by_username(session['username'])
|
||||
|
||||
# Verify current password
|
||||
if not verify_password(current_password, user['password_hash']):
|
||||
flash('Current password is incorrect.', 'error')
|
||||
return redirect(url_for('auth.profile'))
|
||||
|
||||
# Check new password
|
||||
if new_password != confirm_password:
|
||||
flash('New passwords do not match.', 'error')
|
||||
return redirect(url_for('auth.profile'))
|
||||
|
||||
if len(new_password) < 8:
|
||||
flash('Password must be at least 8 characters long.', 'error')
|
||||
return redirect(url_for('auth.profile'))
|
||||
|
||||
# Update password
|
||||
new_hash = hash_password(new_password)
|
||||
with get_db_connection() as conn:
|
||||
with get_db_cursor(conn) as cur:
|
||||
cur.execute("UPDATE users SET password_hash = %s, last_password_change = %s WHERE id = %s",
|
||||
(new_hash, datetime.now(), user['id']))
|
||||
|
||||
log_audit('PASSWORD_CHANGE', 'user', entity_id=user['id'],
|
||||
additional_info="Password changed successfully")
|
||||
flash('Password changed successfully.', 'success')
|
||||
return redirect(url_for('auth.profile'))
|
||||
|
||||
|
||||
@auth_bp.route("/profile/setup-2fa")
|
||||
@login_required
|
||||
def setup_2fa():
|
||||
user = get_user_by_username(session['username'])
|
||||
|
||||
if user['totp_enabled']:
|
||||
flash('2FA is already enabled for your account.', 'info')
|
||||
return redirect(url_for('auth.profile'))
|
||||
|
||||
# Generate new TOTP secret
|
||||
totp_secret = generate_totp_secret()
|
||||
session['temp_totp_secret'] = totp_secret
|
||||
|
||||
# Generate QR code
|
||||
qr_code = generate_qr_code(user['username'], totp_secret)
|
||||
|
||||
return render_template('setup_2fa.html',
|
||||
totp_secret=totp_secret,
|
||||
qr_code=qr_code)
|
||||
|
||||
|
||||
@auth_bp.route("/profile/enable-2fa", methods=["POST"])
|
||||
@login_required
|
||||
def enable_2fa():
|
||||
token = request.form.get('token', '').replace(' ', '')
|
||||
totp_secret = session.get('temp_totp_secret')
|
||||
|
||||
if not totp_secret:
|
||||
flash('2FA setup session expired. Please try again.', 'error')
|
||||
return redirect(url_for('auth.setup_2fa'))
|
||||
|
||||
# Verify the token
|
||||
if not verify_totp(totp_secret, token):
|
||||
flash('Invalid authentication code. Please try again.', 'error')
|
||||
return redirect(url_for('auth.setup_2fa'))
|
||||
|
||||
# Generate backup codes
|
||||
backup_codes = generate_backup_codes()
|
||||
backup_codes_hashed = [hash_backup_code(code) for code in backup_codes]
|
||||
|
||||
# Enable 2FA for user
|
||||
user = get_user_by_username(session['username'])
|
||||
with get_db_connection() as conn:
|
||||
with get_db_cursor(conn) as cur:
|
||||
cur.execute("""
|
||||
UPDATE users
|
||||
SET totp_secret = %s, totp_enabled = true, backup_codes = %s
|
||||
WHERE id = %s
|
||||
""", (totp_secret, json.dumps(backup_codes_hashed), user['id']))
|
||||
|
||||
# Clear temp secret
|
||||
session.pop('temp_totp_secret', None)
|
||||
|
||||
log_audit('2FA_ENABLED', 'user', entity_id=user['id'],
|
||||
additional_info="2FA successfully enabled")
|
||||
|
||||
# Show backup codes
|
||||
return render_template('backup_codes.html', backup_codes=backup_codes)
|
||||
|
||||
|
||||
@auth_bp.route("/profile/disable-2fa", methods=["POST"])
|
||||
@login_required
|
||||
def disable_2fa():
|
||||
password = request.form.get('password')
|
||||
|
||||
user = get_user_by_username(session['username'])
|
||||
|
||||
# Verify password
|
||||
if not verify_password(password, user['password_hash']):
|
||||
flash('Incorrect password. 2FA was not disabled.', 'error')
|
||||
return redirect(url_for('auth.profile'))
|
||||
|
||||
# Disable 2FA
|
||||
with get_db_connection() as conn:
|
||||
with get_db_cursor(conn) as cur:
|
||||
cur.execute("""
|
||||
UPDATE users
|
||||
SET totp_enabled = false, totp_secret = NULL, backup_codes = NULL
|
||||
WHERE id = %s
|
||||
""", (user['id'],))
|
||||
|
||||
log_audit('2FA_DISABLED', 'user', entity_id=user['id'],
|
||||
additional_info="2FA disabled by user")
|
||||
flash('2FA has been disabled for your account.', 'success')
|
||||
return redirect(url_for('auth.profile'))
|
||||
|
||||
|
||||
@auth_bp.route("/heartbeat", methods=['POST'])
|
||||
@login_required
|
||||
def heartbeat():
|
||||
"""Endpoint für Session Keep-Alive - aktualisiert last_activity"""
|
||||
# Aktualisiere last_activity nur wenn explizit angefordert
|
||||
session['last_activity'] = datetime.now(ZoneInfo("Europe/Berlin")).replace(tzinfo=None).isoformat()
|
||||
# Force session save
|
||||
session.modified = True
|
||||
|
||||
return jsonify({
|
||||
'status': 'ok',
|
||||
'last_activity': session['last_activity'],
|
||||
'username': session.get('username')
|
||||
})
|
||||
In neuem Issue referenzieren
Einen Benutzer sperren