ethiele/signage
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Info Tab und MFA

Browse Source
This commit is contained in:
Erik Thiele
2026-06-28 15:14:06 +02:00
parent 1243910d4b
commit 297b52e50c
15 changed files with 5203 additions and 6363 deletions
Download Patch File Download Diff File Expand all files Collapse all files

244
app.py
View File

@@ -13,8 +13,14 @@ import json
import hashlib
import re
import time
import io
import base64
import secrets
import generate_welcome_page # Logo-Suche und Willkommensseite-Generierung
import pyotp
import qrcode
from functools import wraps
from datetime import datetime
from pathlib import Path
@@ -22,7 +28,7 @@ from pathlib import Path
from flask import (
Flask, jsonify, render_template,
send_from_directory, redirect,
request, abort
request, abort, flash, session
)
from flask_login import (
LoginManager, login_user, current_user,
@@ -42,12 +48,35 @@ CONFIG_FILE = os.path.join(BASE_DIR, "config.json") # Persistente Konfiguration
USERS_FILE = os.path.join(BASE_DIR, "users.json") # User-Datenbank
HISTORY_FILE = os.path.join(BASE_DIR, "history.json") # Aktivitätsverlauf
APP_VERSION = "6.0.3"
APP_VERSION = "6.1.0"
UPLOAD_EXTENSIONS = {".jpg", ".jpeg", ".png", ".mp4"}
app = Flask(__name__)
app.secret_key = "CHANGE_THIS_SECRET!!!"
@app.context_processor
def inject_globals():
"""Stellt globale Template-Variablen für Header/Footer bereit."""
import os as _os
hostname = _os.uname().nodename
cfg = load_config()
from flask_login import current_user
if current_user.is_authenticated:
site_list = get_accessible_sites(cfg, current_user)
current_site_val = get_site_list(cfg)[0] if get_site_list(cfg) else ""
else:
site_list = get_site_list(cfg)
current_site_val = site_list[0] if site_list else ""
return dict(
site_list=site_list,
current_site=current_site_val,
brand_subtitle="",
site_dropdown_url_prefix="/admin/",
version=APP_VERSION,
year=datetime.now().year,
hostname=hostname
)
login_manager = LoginManager(app)
login_manager.login_view = "login" # Nicht eingeloggte User werden zum Login umgeleitet
@@ -76,21 +105,38 @@ HEARTBEAT_DIR = Path("/tmp/signage-heartbeat")
HEARTBEAT_TIMEOUT = 60 # Sekunden
def record_heartbeat(site, screen):
"""Aktualisiert den Heartbeat-Timestamp für einen Screen."""
path = HEARTBEAT_DIR / site / screen
def record_heartbeat(site, screen, ip="", ua="", resolution=""):
"""Aktualisiert den Heartbeat-Timestamp für einen Screen und speichert Client-Info."""
path = HEARTBEAT_DIR / site / f"{screen}.json"
path.parent.mkdir(parents=True, exist_ok=True)
path.touch()
data = {
"ip": ip,
"ua": ua[:200] if ua else "",
"resolution": resolution,
"last_seen": datetime.now().strftime("%d.%m.%Y %H:%M:%S")
}
path.write_text(json.dumps(data))
def screen_is_active(site, screen):
"""Prüft, ob ein Screen zuletzt innerhalb des Timeouts registriert wurde."""
path = HEARTBEAT_DIR / site / screen
path = HEARTBEAT_DIR / site / f"{screen}.json"
if not path.exists():
return False
return (time.time() - path.stat().st_mtime) < HEARTBEAT_TIMEOUT
def get_client_info(site, screen):
"""Gibt die gespeicherten Client-Informationen eines Screens zurück."""
path = HEARTBEAT_DIR / site / f"{screen}.json"
if not path.exists():
return {}
try:
return json.loads(path.read_text())
except (json.JSONDecodeError, OSError):
return {}
def get_site_list(cfg):
"""Gibt eine sortierte Liste aller konfigurierten Standort-Namen zurück."""
return sorted(cfg.get("sites", {}).keys())
@@ -293,7 +339,8 @@ class User(UserMixin):
def __init__(self, email, password_hash="", role="user", sites=None,
must_change_password=False, first_name="", last_name="",
department="", notes=""):
department="", notes="", mfa_enabled=False, mfa_secret="",
mfa_recovery_codes=None):
self.email = email
self.id = email
self.password_hash = password_hash
@@ -304,6 +351,9 @@ class User(UserMixin):
self.last_name = last_name
self.department = department
self.notes = notes
self.mfa_enabled = mfa_enabled
self.mfa_secret = mfa_secret
self.mfa_recovery_codes = mfa_recovery_codes or []
@property
def is_admin(self):
@@ -425,6 +475,7 @@ def login():
if request.method == "POST":
email = request.form.get("username", "").strip().lower()
password = request.form.get("password", "")
mfa_code = request.form.get("mfa_code", "")
user = get_user(email)
if user:
@@ -438,11 +489,36 @@ def login():
user.must_change_password = True
if valid:
login_user(user)
add_history_entry("user_login", f"User '{email}' angemeldet", email)
if user.must_change_password:
return redirect("/change-password")
return redirect("/admin")
if user.mfa_enabled:
if mfa_code:
totp = pyotp.TOTP(user.mfa_secret)
if totp.verify(mfa_code):
login_user(user)
add_history_entry("user_login", f"User '{email}' angemeldet (mit MFA)", email)
if user.must_change_password:
return redirect("/change-password")
return redirect("/admin")
# Recovery-Code versuchen
if mfa_code in user.mfa_recovery_codes:
codes = user.mfa_recovery_codes[:]
codes.remove(mfa_code)
users_data = load_users()
users_data[email]["mfa_recovery_codes"] = codes
save_users(users_data)
login_user(user)
add_history_entry("user_login", f"User '{email}' angemeldet (mit Recovery-Code)", email)
return redirect("/mfa/setup?recovered=1")
error = "Ungültiger Sicherheitscode"
else:
session["mfa_user"] = email
session["mfa_password_ok"] = True
return redirect("/mfa/verify")
else:
login_user(user)
add_history_entry("user_login", f"User '{email}' angemeldet", email)
if user.must_change_password:
return redirect("/change-password")
return redirect("/admin")
elif not user.password_hash:
login_user(user)
add_history_entry("user_login", f"User '{email}' angemeldet (Passwort-Änderung erforderlich)", email)
@@ -461,6 +537,134 @@ def logout():
return redirect("/login")
# -------------------------------------------------
# MFA (Multi-Faktor-Authentifizierung per TOTP)
# -------------------------------------------------
@app.route("/mfa/verify", methods=["GET", "POST"])
def mfa_verify():
"""TOTP-Code nach erfolgreichem Login (Passwort) eingeben."""
email = session.get("mfa_user")
if not email:
return redirect("/login")
if request.method == "POST":
code = request.form.get("code", "").strip()
action = request.form.get("action", "")
user = get_user(email)
if not user:
session.pop("mfa_user", None)
return redirect("/login")
if action == "recovery":
return render_template("login.html", error=None, show_mfa=True, mfa_user=email, recovery_mode=True)
totp = pyotp.TOTP(user.mfa_secret)
if code and totp.verify(code):
login_user(user)
session.pop("mfa_user", None)
add_history_entry("user_login", f"User '{email}' angemeldet (mit MFA)", email)
if user.must_change_password:
return redirect("/change-password")
return redirect("/admin")
# Recovery-Code prüfen
if code in user.mfa_recovery_codes:
codes = user.mfa_recovery_codes[:]
codes.remove(code)
users_data = load_users()
users_data[email]["mfa_recovery_codes"] = codes
save_users(users_data)
login_user(user)
session.pop("mfa_user", None)
add_history_entry("user_login", f"User '{email}' angemeldet (mit Recovery-Code)", email)
return redirect("/mfa/setup?recovered=1")
return render_template("mfa_verify.html", error="Ungültiger Code", email=email)
return render_template("mfa_verify.html", error=None, email=email)
@app.route("/mfa/setup", methods=["GET", "POST"])
@login_required
def mfa_setup():
"""MFA einrichten: QR-Code anzeigen, ersten Code verifizieren."""
cfg = load_config()
site_list = get_accessible_sites(cfg, current_user)
email = current_user.email
users_data = load_users()
user_data = users_data.get(email, {})
recovered = request.args.get("recovered")
disabled = request.args.get("disabled")
error_param = request.args.get("error")
secret = user_data.get("mfa_secret") or pyotp.random_base32()
# Secret sofort persistieren, damit GET (QR-Code) und POST (Verifikation) dasselbe Secret verwenden
if not user_data.get("mfa_secret"):
users_data.setdefault(email, {})["mfa_secret"] = secret
save_users(users_data)
totp = pyotp.TOTP(secret)
if request.method == "POST":
code = request.form.get("code", "").strip()
if totp.verify(code):
recovery = [secrets.token_hex(3).upper()[:7] for _ in range(8)]
users_data[email]["mfa_enabled"] = True
users_data[email]["mfa_recovery_codes"] = recovery
save_users(users_data)
add_history_entry("mfa_enabled", f"User '{email}' hat MFA aktiviert", email)
return render_template("mfa_recovery.html", codes=recovery, email=email,
site_list=site_list, current_site=site_list[0] if site_list else "",
brand_subtitle="MFA", site_dropdown_url_prefix="/admin/")
return render_template("mfa_setup.html", error="Code ungültig. Bitte versuche es erneut.",
email=email, secret=secret, qr_data=None,
site_list=site_list, current_site=site_list[0] if site_list else "",
brand_subtitle="MFA", site_dropdown_url_prefix="/admin/")
uri = totp.provisioning_uri(name=email, issuer_name="CANCOM Signage")
img = qrcode.make(uri)
buf = io.BytesIO()
img.save(buf, format="PNG")
qr_b64 = base64.b64encode(buf.getvalue()).decode()
return render_template("mfa_setup.html", error=None if not error_param else "Passwort ist falsch",
email=email, secret=secret, qr_data=qr_b64,
recovered=recovered, disabled=disabled,
site_list=site_list, current_site=site_list[0] if site_list else "",
brand_subtitle="MFA", site_dropdown_url_prefix="/admin/")
@app.route("/mfa/disable", methods=["POST"])
@login_required
def mfa_disable():
"""MFA deaktivieren (mit Passwort-Bestätigung)."""
email = current_user.email
password = request.form.get("password", "")
user = get_user(email)
if user and user.password_hash:
try:
valid = check_password_hash(user.password_hash, password)
except (ValueError, TypeError):
valid = False
else:
valid = False
if not valid:
return redirect("/mfa/setup?error=1")
users_data = load_users()
users_data[email]["mfa_enabled"] = False
users_data[email]["mfa_secret"] = ""
users_data[email]["mfa_recovery_codes"] = []
save_users(users_data)
add_history_entry("mfa_disabled", f"User '{email}' hat MFA deaktiviert", email)
return redirect("/mfa/setup?disabled=1")
# -------------------------------------------------
# Willkommensseite (Kunden-Logos + Generierung)
# -------------------------------------------------
@@ -758,8 +962,12 @@ def player(site, screen):
def playlist_hash(site, screen):
"""Berechnet einen MD5-Hash über die aktuelle Konfiguration (Screen + Priority).
Gibt JSON mit Hash und aktuellem stay_on_first-Wert zurück.
Zeichnet gleichzeitig den Heartbeat auf (Player aktiv)."""
record_heartbeat(site, screen)
Zeichnet gleichzeitig den Heartbeat auf (Player aktiv).
Nimmt optionale Query-Parameter w (Breite) und h (Höhe) für Client-Info."""
ip = request.remote_addr or ""
ua = request.headers.get("User-Agent", "")
resolution = f"{request.args.get('w', '?')}x{request.args.get('h', '?')}"
record_heartbeat(site, screen, ip=ip, ua=ua, resolution=resolution)
config = load_config()
relevant = {
@@ -858,6 +1066,7 @@ def admin(site):
screens = {}
media_files = {}
screen_status = {}
client_info = {}
site_media_base = os.path.join(MEDIA_DIR, current_site)
@@ -944,6 +1153,7 @@ def admin(site):
screens[screen_name] = screen_cfg
media_files[screen_name] = files
screen_status[screen_name] = "online" if screen_is_active(current_site, screen_name) else "offline"
client_info[screen_name] = get_client_info(current_site, screen_name)
return render_template(
"admin.html",
@@ -952,6 +1162,7 @@ def admin(site):
screens=screens,
media_files=media_files,
screen_status=screen_status,
client_info=client_info,
version=APP_VERSION,
year=datetime.now().year,
hostname=os.uname().nodename,
@@ -1427,7 +1638,8 @@ def admin_users():
"first_name": data.get("first_name", ""),
"last_name": data.get("last_name", ""),
"department": data.get("department", ""),
"notes": data.get("notes", "")
"notes": data.get("notes", ""),
"mfa_enabled": data.get("mfa_enabled", False)
})
return render_template(