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

Userverwaltung hinzugefügt

Browse Source
This commit is contained in:
Erik Thiele
2026-06-20 12:09:02 +02:00
parent 2a29025386
commit 39aed78b91
12 changed files with 3252 additions and 2498 deletions
Download Patch File Download Diff File Expand all files Collapse all files

418
app.py
View File

@@ -11,8 +11,10 @@ zugeschaltet werden, die auf allen Playern abwechselnd eingeblendet wird.
import os
import json
import hashlib
import re
import generate_welcome_page # Logo-Suche und Willkommensseite-Generierung
from functools import wraps
from datetime import datetime
from flask import (
@@ -21,11 +23,12 @@ from flask import (
request, abort
)
from flask_login import (
LoginManager, login_user,
LoginManager, login_user, current_user,
login_required, logout_user,
UserMixin
)
from werkzeug.utils import secure_filename
from werkzeug.security import generate_password_hash, check_password_hash
# -------------------------------------------------
@@ -34,8 +37,9 @@ from werkzeug.utils import secure_filename
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
MEDIA_DIR = os.path.join(BASE_DIR, "media") # Medien-Dateien je Standort/Screen
CONFIG_FILE = os.path.join(BASE_DIR, "config.json") # Persistente Konfiguration
USERS_FILE = os.path.join(BASE_DIR, "users.json") # User-Datenbank
APP_VERSION = "5.5.0"
APP_VERSION = "6.0.1"
UPLOAD_EXTENSIONS = {".jpg", ".jpeg", ".png", ".mp4"}
app = Flask(__name__)
@@ -224,17 +228,112 @@ def media_background(site, filename):
# -------------------------------------------------
# User & Login (Flask-Login)
# User-Verwaltung (Flask-Login mit users.json)
# -------------------------------------------------
class Admin(UserMixin):
"""Einfacher Admin-User. Es gibt nur einen Admin (keine DB)."""
id = 1
class User(UserMixin):
"""User mit E-Mail (id), Rolle und Standort-Berechtigungen."""
def __init__(self, email, password_hash="", role="user", sites=None, must_change_password=False):
self.email = email
self.id = email
self.password_hash = password_hash
self.role = role
self.sites = sites or []
self.must_change_password = must_change_password
@property
def is_admin(self):
return self.role == "admin"
@property
def is_superuser(self):
return self.role == "superuser"
def can_access_site(self, site):
return self.is_admin or self.is_superuser or site in self.sites
def load_users():
"""Liest die users.json. Gibt leeres Dict zurück, wenn nicht vorhanden."""
if not os.path.exists(USERS_FILE):
return {}
with open(USERS_FILE) as f:
return json.load(f)
def save_users(users):
"""Schreibt das User-Dict zurück in die users.json."""
with open(USERS_FILE, "w") as f:
json.dump(users, f, indent=2)
def get_user(email):
"""Lädt einen einzelnen User aus users.json. Gibt None zurück, wenn nicht gefunden."""
users = load_users()
data = users.get(email)
if data:
return User(email, **data)
return None
def init_user_db():
"""Legt users.json mit einem Admin-Konto aus config.json.admin an, falls sie nicht existiert."""
if os.path.exists(USERS_FILE):
return
cfg = load_config()
admin = cfg.get("admin", {})
username = admin.get("username", "admin@signage.local")
password = admin.get("password", "admin")
users = {
username: {
"password_hash": generate_password_hash(password),
"role": "admin",
"sites": [],
"must_change_password": False
}
}
save_users(users)
print(f"✅ users.json mit Admin-Konto ({username}) angelegt (Passwort aus config.json)")
@login_manager.user_loader
def load_user(user_id):
"""Lädt den Admin-User für Flask-Login. user_id wird ignoriert – es gibt nur einen User."""
return Admin()
"""Lädt einen User aus users.json anhand der E-Mail (Flask-Login-Callback)."""
return get_user(user_id)
# -------------------------------------------------
# Dekoratoren: Admin / Standort-Zugriff
# -------------------------------------------------
def admin_required(f):
"""Erfordert einen Admin-User (role=admin)."""
@wraps(f)
@login_required
def decorated_function(*args, **kwargs):
if not current_user.is_admin:
abort(403)
return f(*args, **kwargs)
return decorated_function
def site_access_required(f):
"""Erfordert Zugriff auf den Standort (admin →alle, user →nur freigegebene)."""
@wraps(f)
@login_required
def decorated_function(*args, **kwargs):
site = kwargs.get("site")
if site and not current_user.can_access_site(site):
abort(403)
return f(*args, **kwargs)
return decorated_function
def get_accessible_sites(cfg, user):
"""Gibt die Liste der Standorte zurück, auf die ein User Zugriff hat."""
all_sites = sorted(cfg.get("sites", {}).keys())
if user.is_admin or user.is_superuser:
return all_sites
return [s for s in all_sites if s in user.sites]
# -------------------------------------------------
@@ -242,20 +341,36 @@ def load_user(user_id):
# -------------------------------------------------
@app.route("/login", methods=["GET", "POST"])
def login():
"""Login-Seite. Bei POST werden die Zugangsdaten aus config.json.admin geprüft."""
config = load_config()
"""Login-Seite. Prüft Zugangsdaten gegen users.json (gehashte Passwörter)."""
error = None
next_url = None
if request.method == "POST":
if (
request.form.get("username") == config["admin"].get("username")
and request.form.get("password") == config["admin"].get("password")
):
login_user(Admin())
return redirect("/admin")
email = request.form.get("username", "").strip().lower()
password = request.form.get("password", "")
user = get_user(email)
if user:
if user.password_hash:
try:
valid = check_password_hash(user.password_hash, password)
except (ValueError, TypeError):
valid = False
else:
valid = False
user.must_change_password = True
if valid:
login_user(user)
if user.must_change_password:
return redirect("/change-password")
return redirect("/admin")
elif not user.password_hash:
login_user(user)
return redirect("/change-password")
error = "Ungültige Zugangsdaten"
return render_template("login.html", error=error)
return render_template("login.html", error=error, next=next_url)
@app.route("/logout")
@@ -270,6 +385,7 @@ def logout():
# -------------------------------------------------
@app.route("/willkommen", methods=["GET", "POST"])
@app.route("/customer", methods=["GET", "POST"])
@login_required
def add_customer():
"""
Formular für die Willkommensseite:
@@ -283,12 +399,16 @@ def add_customer():
customer_names = ["", "", ""]
site = request.args.get("site") or request.form.get("site") or ""
# Fallback auf ersten Standort, wenn keiner angegeben
# Fallback auf ersten für den User zugänglichen Standort
if not site:
config = load_config()
sites = get_site_list(config)
sites = get_accessible_sites(config, current_user)
site = sites[0] if sites else ""
# Prüfen, ob der User Zugriff auf den Standort hat
if site and not current_user.can_access_site(site):
abort(403)
if request.method == "POST":
customer_names = [
request.form.get("customer_name_1", "").strip(),
@@ -582,18 +702,18 @@ def root_redirect():
@app.route("/admin", strict_slashes=False)
@login_required
def admin_redirect():
"""Leitet zum ersten konfigurierten Standort weiter."""
"""Leitet zum ersten für den User zugänglichen Standort weiter."""
cfg = load_config()
sites = get_site_list(cfg)
sites = get_accessible_sites(cfg, current_user)
if sites:
return redirect(f"/admin/{sites[0]}")
return "<h1>Kein Standort konfiguriert</h1><p>Bitte lege einen Standort im media-Verzeichnis an.</p>", 200
@app.route("/add-site")
@login_required
@admin_required
def add_site():
"""Legt einen neuen Standort an (Verzeichnis + Config-Eintrag)."""
"""Legt einen neuen Standort an (Verzeichnis + Config-Eintrag). Nur für Admins."""
name = request.args.get("name", "").strip().lower()
if not name:
return redirect("/admin")
@@ -607,9 +727,9 @@ def add_site():
@app.route("/admin/<site>/delete-site", methods=["POST"])
@login_required
@admin_required
def delete_site(site):
"""Löscht einen Standort inkl. aller Screens und Medien (Config + Verzeichnis)."""
"""Löscht einen Standort inkl. aller Screens und Medien (Config + Verzeichnis). Nur für Admins."""
cfg = load_config()
cfg.get("sites", {}).pop(site, None)
save_config(cfg)
@@ -624,7 +744,7 @@ def delete_site(site):
@app.route("/admin/<site>", strict_slashes=False)
@login_required
@site_access_required
def admin(site):
"""
Admin-Dashboard für einen Standort.
@@ -632,10 +752,10 @@ def admin(site):
"""
cfg = load_config()
site_list = get_site_list(cfg)
site_list = get_accessible_sites(cfg, current_user)
# Prüfen, ob der Standort existiert
if site not in site_list:
if site not in get_site_list(cfg):
if os.path.isdir(os.path.join(MEDIA_DIR, site)):
cfg.setdefault("sites", {})[site] = {"screens": {}}
save_config(cfg)
@@ -753,11 +873,11 @@ def admin(site):
# Admin: Priority-Playlist (eigene Seite)
# -------------------------------------------------
@app.route("/admin/<site>/priority")
@login_required
@site_access_required
def admin_priority(site):
"""Zeigt die Priority-Playlist-Seite mit Tabs (Playlist + Medien)."""
cfg = load_config()
site_list = sorted(cfg.get("sites", {}).keys())
site_list = get_accessible_sites(cfg, current_user)
priority_files = load_priority_files()
return render_template(
@@ -776,7 +896,7 @@ def admin_priority(site):
# Admin: Hintergrundbild für Willkommensseite hochladen
# -------------------------------------------------
@app.route("/admin/<site>/upload-background", methods=["POST"])
@login_required
@site_access_required
def upload_background(site):
"""Lädt ein Hintergrundbild für die Willkommensseite hoch (pro Standort).
Gespeichert als media/<site>/background.<ext>."""
@@ -806,7 +926,7 @@ def upload_background(site):
@app.route("/admin/<site>/delete-background", methods=["POST"])
@login_required
@site_access_required
def delete_background(site):
"""Löscht das Hintergrundbild für die Willkommensseite."""
site_dir = os.path.join(MEDIA_DIR, site)
@@ -825,7 +945,7 @@ def delete_background(site):
# Admin: Screen-Einstellungen speichern
# -------------------------------------------------
@app.route("/admin/<site>/update/<screen>", methods=["POST"])
@login_required
@site_access_required
def update_screen(site, screen):
"""Speichert die allgemeinen Einstellungen für einen Screen (Intervall, Newsticker, Show-Bilder/Videos)."""
config = load_config()
@@ -843,7 +963,7 @@ def update_screen(site, screen):
@app.route("/admin/<site>/update-actions/<screen>", methods=["POST"])
@login_required
@site_access_required
def update_screen_actions(site, screen):
"""Speichert die Aktionen-Einstellungen für einen Screen (Custom-URL + Position)."""
config = load_config()
@@ -863,7 +983,7 @@ def update_screen_actions(site, screen):
# Admin: Voice-Agent-Einstellungen speichern
# -------------------------------------------------
@app.route("/admin/<site>/update-voice/<screen>", methods=["POST"])
@login_required
@site_access_required
def update_screen_voice(site, screen):
"""Speichert die Voice-Agent-Einstellungen für einen Screen (Enabled, Label, Target, Position)."""
config = load_config()
@@ -883,7 +1003,7 @@ def update_screen_voice(site, screen):
# Admin: Screen löschen
# -------------------------------------------------
@app.route("/admin/<site>/delete-screen/<screen>", methods=["POST"])
@login_required
@site_access_required
def delete_screen(site, screen):
"""Löscht einen Screen (Config-Eintrag + Medien-Verzeichnis)."""
config = load_config()
@@ -904,7 +1024,7 @@ def delete_screen(site, screen):
# Admin: Neuen Screen anlegen
# -------------------------------------------------
@app.route("/admin/<site>/add-screen", methods=["GET"])
@login_required
@site_access_required
def add_screen(site):
"""Legt einen neuen Screen an (Verzeichnis + Config-Eintrag)."""
screen_name = request.args.get("name", "").strip().lower()
@@ -927,7 +1047,7 @@ def add_screen(site):
# Admin: Medien-Upload
# -------------------------------------------------
@app.route("/admin/<site>/upload/<screen>", methods=["POST"])
@login_required
@site_access_required
def upload(site, screen):
"""Lädt eine Datei hoch (Bild oder Video). Bei screen="priority" ins globale Verzeichnis."""
file = request.files.get("file")
@@ -952,7 +1072,7 @@ def upload(site, screen):
# Admin: URL zur Playlist hinzufügen
# -------------------------------------------------
@app.route("/admin/<site>/add-url/<screen>", methods=["POST"])
@login_required
@site_access_required
def add_url(site, screen):
"""Fügt eine URL (mit Zoom-Faktor) zur Playlist eines Screens oder der Priority-Playlist hinzu."""
url = request.form.get("url", "").strip()
@@ -985,7 +1105,7 @@ def add_url(site, screen):
# Admin: Datei aus Playlist löschen
# -------------------------------------------------
@app.route("/admin/<site>/delete/<screen>/<path:filename>", methods=["POST"])
@login_required
@site_access_required
def delete_file(site, screen, filename):
"""Löscht eine Datei aus der Playlist (Config + Dateisystem)."""
cfg = load_config()
@@ -1017,7 +1137,7 @@ def delete_file(site, screen, filename):
# Admin: Playlist-Reihenfolge speichern (Drag & Drop)
# -------------------------------------------------
@app.route("/admin/<site>/playlist/<screen>", methods=["POST"])
@login_required
@site_access_required
def save_playlist(site, screen):
"""
Speichert die über SortableJS per Drag & Drop neu sortierte Playlist.
@@ -1058,8 +1178,224 @@ def save_playlist(site, screen):
return "", 204
# -------------------------------------------------
# Change-Password (für First-Login / Reset)
# -------------------------------------------------
@app.route("/change-password", methods=["GET", "POST"])
@login_required
def change_password():
"""Seite zum Ändern des Passworts (bei First-Login oder nach Admin-Reset)."""
cfg = load_config()
site_list = get_accessible_sites(cfg, current_user)
current_site = site_list[0] if site_list else ""
error = None
success = None
if request.method == "POST":
new_pw = request.form.get("new_password", "")
confirm = request.form.get("confirm_password", "")
if not new_pw or len(new_pw) < 6:
error = "Passwort muss mindestens 6 Zeichen lang sein"
elif new_pw != confirm:
error = "Passwörter stimmen nicht überein"
else:
users = load_users()
email = current_user.email
if email in users:
users[email]["password_hash"] = generate_password_hash(new_pw)
users[email]["must_change_password"] = False
save_users(users)
return redirect("/admin")
return render_template(
"change_password.html",
error=error,
success=success,
site_list=site_list,
current_site=current_site,
version=APP_VERSION,
year=datetime.now().year,
hostname=os.uname().nodename,
server_url=cfg.get("server_url", "")
)
# -------------------------------------------------
# User-Verwaltung (nur für Admins)
# -------------------------------------------------
@app.route("/admin/users")
@admin_required
def admin_users():
"""Listet alle User."""
cfg = load_config()
users_data = load_users()
users_list = []
for email, data in users_data.items():
users_list.append({
"email": email,
"role": data.get("role", "user"),
"sites": data.get("sites", []),
"must_change_password": data.get("must_change_password", False)
})
return render_template(
"user_list.html",
users=sorted(users_list, key=lambda u: u["email"]),
site_list=get_accessible_sites(cfg, current_user),
current_site=get_site_list(cfg)[0] if get_site_list(cfg) else "",
version=APP_VERSION,
year=datetime.now().year,
hostname=os.uname().nodename,
server_url=cfg.get("server_url", "")
)
@app.route("/admin/users/create", methods=["GET"])
@admin_required
def admin_users_create_page():
"""Zeigt das Formular zum Anlegen eines neuen Users."""
cfg = load_config()
all_sites = get_site_list(cfg)
return render_template(
"user_create.html",
all_sites=all_sites,
site_list=get_accessible_sites(cfg, current_user),
current_site=all_sites[0] if all_sites else "",
version=APP_VERSION,
year=datetime.now().year,
hostname=os.uname().nodename,
server_url=cfg.get("server_url", "")
)
@app.route("/admin/users/create", methods=["POST"])
@admin_required
def admin_users_create():
"""Legt einen neuen User an (E-Mail=Username, Passwort, Rolle, Standorte)."""
email = request.form.get("email", "").strip().lower()
password = request.form.get("password", "")
role = request.form.get("role", "user")
sites = request.form.getlist("sites")
# Validierung
if not email or not re.match(r"[^@]+@[^@]+\.[^@]+", email):
return "Ungültige E-Mail-Adresse", 400
if not password or len(password) < 6:
return "Passwort muss mindestens 6 Zeichen lang sein", 400
if role not in ("admin", "superuser", "user"):
return "Ungültige Rolle", 400
users = load_users()
if email in users:
return "User existiert bereits", 400
users[email] = {
"password_hash": generate_password_hash(password),
"role": role,
"sites": sites if role == "user" else [],
"must_change_password": True
}
save_users(users)
return redirect("/admin/users")
@app.route("/admin/users/edit/<path:email>", methods=["GET", "POST"])
@admin_required
def admin_users_edit(email):
"""Bearbeitet einen User (E-Mail, Rolle, Standorte)."""
cfg = load_config()
all_sites = get_site_list(cfg)
users_data = load_users()
if email not in users_data:
return "User nicht gefunden", 404
error = None
if request.method == "POST":
new_email = request.form.get("email", "").strip().lower()
new_role = request.form.get("role", "user")
new_sites = request.form.getlist("sites")
if not new_email or not re.match(r"[^@]+@[^@]+\.[^@]+", new_email):
error = "Ungültige E-Mail-Adresse"
elif new_role not in ("admin", "superuser", "user"):
error = "Ungültige Rolle"
else:
user_data = users_data.pop(email)
user_data["role"] = new_role
user_data["sites"] = new_sites if new_role == "user" else []
users_data[new_email] = user_data
save_users(users_data)
return redirect(f"/admin/users?updated={new_email}")
user = {
"email": email,
"role": users_data[email].get("role", "user"),
"sites": users_data[email].get("sites", []),
"must_change_password": users_data[email].get("must_change_password", False)
}
return render_template(
"user_edit.html",
user=user,
all_sites=all_sites,
error=error,
site_list=get_accessible_sites(cfg, current_user),
current_site=all_sites[0] if all_sites else "",
version=APP_VERSION,
year=datetime.now().year,
hostname=os.uname().nodename,
server_url=cfg.get("server_url", "")
)
@app.route("/admin/users/delete/<path:email>", methods=["POST"])
@admin_required
def admin_users_delete(email):
"""Löscht einen User. Der eigene Account kann nicht gelöscht werden.
Der letzte verbleibende Admin kann nicht gelöscht werden."""
if email == current_user.email:
return "Du kannst dich nicht selbst löschen", 400
users = load_users()
# Prüfen, ob es der letzte Admin wäre
if users.get(email, {}).get("role") == "admin":
admin_count = sum(1 for u in users.values() if u.get("role") == "admin")
if admin_count <= 1:
return "Letzter Admin kann nicht gelöscht werden", 400
users.pop(email, None)
save_users(users)
return redirect("/admin/users")
@app.route("/admin/users/reset-password/<path:email>", methods=["POST"])
@admin_required
def admin_users_reset_password(email):
"""Setzt ein temporäres Passwort und zwingt User zur Änderung beim nächsten Login."""
import secrets
import string
users = load_users()
if email not in users:
return "User nicht gefunden", 404
temp_password = secrets.choice(string.ascii_lowercase) + secrets.choice(string.ascii_uppercase) + secrets.choice(string.digits) + secrets.token_hex(4)
users[email]["password_hash"] = generate_password_hash(temp_password)
users[email]["must_change_password"] = True
save_users(users)
return redirect(f"/admin/users?reset={email}&temp={temp_password}")
# -------------------------------------------------
# Main
# -------------------------------------------------
if __name__ == "__main__":
init_user_db()
app.run(debug=True, host="0.0.0.0", port=5005)