imaginaerraum_door_admin/templates/admins.html

@@ -1,57 +0,0 @@
-{% extends 'base.html' %}
-{% block header %}
-    {% block title %}<h1>Admin Übersicht</h1>{% endblock %}
-
-    <script src="../static/jquery-3.6.0.js"></script>
-{% endblock %}
-
-{% block content %}
-<table border="1">
-<td>Username</td>
-<td>E-Mail</td>
-<td>Aktiv</td>
-<td>Aktionen</td>
-{% for data in admin_data %}
-    <tr>
-    {% for field in ['username', 'email', 'active'] %}
-    <td>{{ data[field] if data[field] }}</td>
-    {% endfor %}
-    <td>
-    <a href="{{ url_for('admin_toggle_active', username=data['username']) }}"><img src="static/stop.png" title="Aktivieren/Deaktivieren" alt="Toggle active"></a>
-    <a href="{{ url_for('delete_admins', username=data['username']) }}"><img src="static/delete.png" title="Löschen" alt="Delete"></a>
-    </td>
-    </tr>
-{% endfor %}
-</table>
-<div>
-    <p>
-    Neuen Admin erstellen:
-    </p>
-    <form method="POST">
-    <table>
-    {{ form.csrf_token }}
-    <tr>
-    <td>{{ form.name.label }}</td>
-    <td>{{ form.name(size=20) }}</td>
-    </tr>
-    <tr>
-    <td>{{ form.email.label }}</td>
-    <td>{{ form.email(size=20) }}</td>
-    </tr>
-    <tr>
-        <td></td>
-        <td>
-            <input type="submit" value="Abschicken">
-        </td>
-    </tr>
-    </table>
-    </form>
-</div>
-<form action="{{ url_for('backup_user_datastore') }}" method="get">
-    <input type="submit" value="Admin Daten sichern">
-</form>
-<form action="{{ url_for('restore_user_datastore') }}" method=post enctype=multipart/form-data>
-  <input type=file name=file>
-  <input type=submit  value="Admin Daten wiederherstellen">
-</form>
-{% endblock %}

imaginaerraum_door_admin/templates/base.html

@@ -8,9 +8,6 @@
     <li><a href="{{ url_for('list_tokens') }}">Token Übersicht</a>
     <li><a href="{{ url_for('open_door') }}">Tür öffnen</a>
     <li><a href="{{ url_for('close_door') }}">Tür schließen</a>
-    {% if current_user.has_role('super_admin') %}
-    <li><a href="{{ url_for('manage_admins') }}">Admins verwalten</a>
-    {% endif %}
     {% if current_user.is_authenticated %}
         <li><a href="{{ url_for('security.change_password') }}">Passwort ändern</a>
         <li><a href="{{ url_for('security.logout') }}">Benutzer <span>{{ current_user.username }}</span> ausloggen</a>

imaginaerraum_door_admin/templates/delete_admin.html

@@ -1,27 +0,0 @@
-{% extends 'base.html' %}
-{% block header %}
-    {% block title %}<h1>Admin Benutzer löschen</h1>{% endblock %}
-    <script src="../static/jquery-3.6.0.js"></script>
-{% endblock %}
-
-{% block content %}
-<div>
-    Achtung, Admin '{{ username }}' wird gelöscht.
-    Bitte zur Bestätigung den Nutzernamen eingeben:
-    <form method="POST">
-    <table>
-    {{ form.csrf_token }}
-    <tr>
-        <td>{{ form.name.label }}</td>
-        <td>{{ form.name(size=20) }}</td>
-    </tr>
-    <tr>
-        <td></td>
-        <td>
-        <input type="submit" value="Bestätigen">
-        </td>
-    </tr>
-    </table>
-    </form>
-</div>
-{% endblock %}

imaginaerraum_door_admin/templates/tokens.html

@@ -39,7 +39,4 @@
     </tr>
 {% endfor %}
 </table>
-<form action="{{ url_for('backup_tokens') }}" method="get">
-    <input type="submit" value="Token Daten sichern">
-</form>
 {% endblock %}

imaginaerraum_door_admin/webapp.py

@@ -1,27 +1,21 @@
 import os
-from flask import Flask, render_template, request, flash, redirect, session, send_file
+from flask import Flask, render_template, request, flash, redirect, session, url_for
-from werkzeug.utils import secure_filename
 from flask_wtf import FlaskForm
 from wtforms.fields.html5 import DateField, EmailField
 from wtforms.fields import StringField, BooleanField
 from wtforms.validators import DataRequired, ValidationError, EqualTo
 from flask_sqlalchemy import SQLAlchemy
-from flask_security import Security, SQLAlchemyUserDatastore, auth_required, hash_password, uia_email_mapper, \
+from flask_security import Security, SQLAlchemyUserDatastore, auth_required, hash_password, uia_email_mapper, current_user
-    current_user, roles_required
 from flask_security.models import fsqla_v2 as fsqla
 from flask_security.forms import LoginForm, Required, PasswordField
 from flask_security.utils import find_user, verify_password
 from flask_mail import Mail
-from email_validator import validate_email, EmailNotValidError
+from email_validator import validate_email
-
-import json
-import secrets
 import bleach
 import ldap3
 from ldap3.core.exceptions import LDAPBindError, LDAPSocketOpenError
 from pathlib import Path
 import logging
-import tempfile
 
 from datetime import date
 from .door_handle import DoorHandle
@@ -47,22 +41,14 @@ class TokenForm(FlaskForm):
     active = BooleanField('Aktiv?')
     dsgvo = BooleanField('Einwilligung Nutzungsbedingungen erfragt?', validators=[DataRequired()])
 
-
+class TokenDeleteForm(FlaskForm):
-class ConfirmDeleteForm(FlaskForm):
     name = StringField('Name', validators=[DataRequired(), EqualTo('name_confirm', 'Name stimmt nicht überein')])
     name_confirm = StringField('Name confirm')
 
-
-class AdminCreationForm(FlaskForm):
-    name = StringField('Name', validators=[DataRequired()])
-    email = EmailField('E-Mail', validators=[DataRequired()])
-
-
 def uia_username_mapper(identity):
     # we allow pretty much anything - but we bleach it.
     return bleach.clean(identity, strip=True)
 
-
 def create_application(config):
     # set up logging for the web app
     logger = logging.getLogger('webapp')
@@ -90,6 +76,22 @@ def create_application(config):
     if not Path(config.token_file).exists():
         logger.warning(f"Token file not found at {config.token_file}")
 
+    new_admin_data = []
+    if config.admin_file is not None:
+        if not Path(config.admin_file).exists():
+            logger.warning(f"Admin user creation file not found at {config.admin_file}")
+        else:
+            # store data for new admins in memory s.t. the file can be deleted afterwards
+            with open(config.admin_file) as f:
+                for i, d in enumerate(f.readlines()):
+                    try:
+                        user, email, pw = d.split()
+                        validate_email(email)
+                        new_admin_data.append({'username': user, 'email': email, 'password': pw})
+                    except Exception as e:
+                        print(f"Error while parsing line {i} in admin config file. Config file should contain lines of "
+                              f"'<username> <email> <password>\\n'\n Exception: {e}\nAdmin account could not be created.")
+
     # create door objects which provides access to the token file and current door state via MQTT
     door = DoorHandle(token_file=config.token_file, mqtt_host=config.mqtt_host, nfc_socket=config.nfc_socket,
                       logger=logger)
@@ -146,7 +148,7 @@ def create_application(config):
     ldap_server = ldap3.Server(config.ldap_url)
     local_ldap_cache = {}   # dict for caching LDAP authorization locally (stores username + hashed password)
 
-    def validate_ldap(username, password):
+    def validate_ldap(user, password):
         """Validate the user and password through an LDAP server.
 
         If the connection completes successfully the given user and password is authorized and the password is stored
@@ -165,7 +167,7 @@ def create_application(config):
         """
 
         try:
-            con = ldap3.Connection(ldap_server, user=f"uid={username},ou=Users,dc=imaginaerraum,dc=de",
+            con = ldap3.Connection(ldap_server, user="uid=%s,ou=Users,dc=imaginaerraum,dc=de" % (user.username,),
                                     password=password, auto_bind=True)
         except ldap3.core.exceptions.LDAPBindError:
             # server reachable but user unauthorized -> fail
@@ -173,14 +175,11 @@ def create_application(config):
         except LDAPSocketOpenError:
             # server not reachable -> try cached authorization data
             return user.username in local_ldap_cache and verify_password(password, local_ldap_cache[user.username])
-        except Exception as e:
+        except Exception:
             # for other Exceptions we just fail
             return False
 
         # TODO check if user has permission to edit tokens
-        lock_permission = con.search('ou=Users,dc=imaginaerraum,dc=de', f'(&(uid={user.username})(memberof=cn=Members,ou=Groups,dc=imaginaerraum,dc=de))')
-        token_granting_permission = con.search('ou=Users,dc=imaginaerraum,dc=de',
-                            f'(&(uid={user.username})(memberof=cn=Vorstand,ou=Groups,dc=imaginaerraum,dc=de))')
         # if LDAP authorization succeeds we cache the password locally (in memory) to allow LDAP authentication even if
         # the server is not reachable
         local_ldap_cache[user.username] = hash_password(password)
@@ -191,32 +190,17 @@ def create_application(config):
         password = PasswordField('Passwort', [Required()])
 
         def validate(self):
-            # try authorizing using LDAP
-            # authorization in LDAP uses username -> get username associated with email from the database
-            try:
-                # if an email (instead of a username) was entered for authentication we check if there already is a user
-                # with that email in the database
-                validate_email(self.email.data)
-                user = find_user(self.email.data)
-                if user is not None:
-                    username = user.username
-                else:
-                    # this means there is no user with that email in the database
-                    username = None
-            except EmailNotValidError:
-                # else we use the entered credentials as username
-                username = self.email.data
-
-            authorized = validate_ldap(username, self.password.data)
-
-            if authorized:
-                logger.info(f"Admin user with credentials '{self.email.data}' authorized through LDAP")
-
-            if not authorized:
             # try authorizing locally using Flask security user datastore
             authorized = super(ExtendedLoginForm, self).validate()
 
+            if not authorized:
+                # try authorizing using LDAP
+                # authorization in LDAP uses username -> get username associated with email from the database
+                user = find_user(self.email.data)
+                authorized = validate_ldap(user, self.password.data)
                 if authorized:
+                    logger.info(f"Admin user with credentials '{self.email.data}' authorized through LDAP")
+            else:
                 logger.info(f"Admin user with credentials '{self.email.data}' authorized through local database")
 
             # if any of the authorization methods is successful we authorize the user
@@ -226,148 +210,28 @@ def create_application(config):
     user_datastore = SQLAlchemyUserDatastore(db, User, Role)
     security = Security(app, user_datastore, login_form=ExtendedLoginForm)
 
-    # admin user management
+    # create admin users (only if they don't exists already)
-    @app.route('/manage_admins', methods=['GET', 'POST'])
+    def create_admins(new_admin_data):
-    @roles_required('super_admin')
+        for d in new_admin_data:
-    def manage_admins():
+            if user_datastore.find_user(email=d['email'], username=d['username']) is None:
-        form = AdminCreationForm()
+                logger.info(f"New admin user created with username '{d['username']}' and email '{d['email']}'")
-        if request.method == 'GET':
+                # create new admin (only if admin does not already exist)
-            users = user_datastore.user_model.query.all()
+                user_datastore.create_user(email=d['email'], username=d['username'], password=hash_password(d['password']))
-            admin_data = [{'username': u.username, 'email': u.email, 'active': u.is_active} for u in users]
+
-            return render_template('admins.html', admin_data=admin_data, form=form)
+    # Create a user to test with
-        elif form.validate():
+    @app.before_first_request
-            if user_datastore.find_user(username=form.name.data) is not None or \
+    def create_user():
-                    user_datastore.find_user(email=form.email.data) is not None:
+        db.create_all()
-                flash("A user with the same name or email is already registered!")
+        if len(new_admin_data) > 0:
-                return redirect('/manage_admins')
+            # create admin accounts from given file
-            else:
+            create_admins(new_admin_data)
-                pw = secrets.token_urlsafe(16)
-                new_user = user_datastore.create_user(username=form.name.data, email=form.email.data, roles=['admin'],
-                                                      password=hash_password(pw))
-                logger.info(
-                    f"Super admin {current_user.username} created new admin account for {new_user.username} <{new_user.email}>")
-                flash(
-                    f"An account for the new admin user {new_user.username} has been created. Use the randomly generated password {pw} to log in.")
         db.session.commit()
-                return redirect('/manage_admins')
 
-    @app.route('/delete_admins/<username>', methods=['GET', 'POST'])
-    @roles_required('super_admin')
-    def delete_admins(username):
-        user = user_datastore.find_user(username=username)
-        if user is None:
-            flash(f"Invalid user {username}")
-            return redirect('/manage_admins')
-        if user.has_role('super_admin'):
-            flash('Cannot delete super admins!')
-            return redirect('/manage_admins')
-        if user.is_active:
-            flash('Cannot delete active users. Please deactivate user first!')
-            return redirect('/manage_admins')
-
-        # set up form for confirming deletion
-        form = ConfirmDeleteForm()
-        form.name_confirm.data = username
-
-        if request.method == 'GET':
-            # return page asking the user to confirm delete
-            return render_template('delete_admin.html', username=username, form=form)
-        elif form.validate():
-            user_datastore.delete_user(user)
-            flash(f"Username {username} was deleted.")
-            logger.info(f"Super admin {current_user.username} deleted admin user {username}")
-            db.session.commit()
-            return redirect('/manage_admins')
-        else:
-            flash("Username does not match. User was not deleted!")
-            return redirect('/manage_admins')
-
-    @app.route('/admin_toggle_active/<username>')
-    @roles_required('super_admin')
-    def admin_toggle_active(username):
-        user = user_datastore.find_user(username=username)
-        if user is None:
-            flash(f"Invalid user {username}")
-            return redirect('/manage_admins')
-        if user.has_role('super_admin'):
-            flash('Cannot deactivate super admins!')
-            return redirect('/manage_admins')
-        user_datastore.toggle_active(user)
-        if user.is_active:
-            logger.info(f"Super admin {current_user.username} activated access for admin user {username}")
-        else:
-            logger.info(f"Super admin {current_user.username} deactivated access for admin user {username}")
-        db.session.commit()
-        return redirect('/manage_admins')
-
-    @app.route('/backup_user_datastore')
-    @roles_required('super_admin')
-    def backup_user_datastore():
-        # get list of defined admin users for backup
-        users = user_datastore.user_model.query.all()
-        user_data = [{'username': u.username, 'email': u.email, 'active': u.is_active, 'password_hash': u.password}
-                     for u in users if not u.has_role('super_admin')]
-        try:
-            with tempfile.TemporaryDirectory() as tmpdir:
-                file = Path(tmpdir, 'user_data.txt')
-                file.write_text(json.dumps(user_data))
-                return send_file(file, as_attachment=True, cache_timeout=-1)
-        except Exception as e:
-            return str(e)
-
-    @app.route('/restore_user_datastore', methods=['POST'])
-    @roles_required('super_admin')
-    def restore_user_datastore():
-        # check if the post request has the file part
-        if 'file' not in request.files:
-            flash('Keine Datei ausgewählt!')
-            return redirect(request.url)
-        file = request.files['file']
-        # if user does not select file, browser also
-        # submit an empty part without filename
-        if file.filename == '':
-            flash('Keine Datei ausgewählt!')
-            return redirect('/manage_admins')
-        filename = secure_filename(file.filename)
-        if file and filename.endswith('.txt'):
-            data = file.stream.read()
-            try:
-                # check validity of user data
-                user_data = json.loads(data)
-                valid = type(user_data) == list
-                valid &= all(type(d) == dict for d in user_data)
-                if valid:
-                    for d in user_data:
-                        entry_valid = set(d.keys()) == { 'active', 'email', 'password_hash', 'username'}
-                        entry_valid &= all(len(d[key]) > 0 for key in ['email', 'password_hash', 'username'])
-                        entry_valid &= type(d['active']) == bool
-                        validate_email(d['email'])
-                        if entry_valid:
-                            existing_user = user_datastore.find_user(username=d['username'], email=d['email'])
-                            if existing_user is None:
-                                user_datastore.create_user(username=d['username'], email=d['email'], password=d['password_hash'],
-                                           roles=['admin'], active=d['active'])
-                                flash(f"Admin Account für Benutzer '{d['username']} wurde wiederhergestellt.")
-                            else:
-                                flash(f"Admin '{d['username']} existiert bereits. Eintrag wird übersprungen.")
-                        else:
-                            raise ValueError(f"Ungültige Daten für User Entry {d}")
-                else:
-                    raise ValueError("Admin User Datei hat ungültiges Format.")
-            except Exception as e:
-                flash(f"Die Datei konnte nicht gelesen werden. Exception: {e}")
-                return redirect('/manage_admins')
-            flash("Admin Benutzer aus Datei gelesen.")
-            db.session.commit()
-        else:
-            flash("Ungültige Dateiendung")
-        return redirect('/manage_admins')
-    # main page
     @app.route('/')
     def door_lock():
         return render_template('index.html', door_state=door.state, encoder_position=door.encoder_position)
 
-    # token overview
+
     @app.route('/tokens')
     @auth_required()
     def list_tokens():
@@ -376,7 +240,7 @@ def create_application(config):
         inactive_tokens = {t: data for t, data in tokens.items() if data['inactive']}
         return render_template('tokens.html', assigned_tokens=assigned_tokens, inactive_tokens=inactive_tokens)
 
-    # routes for registering, editing and deleting tokens
+
     @app.route('/register-token', methods=['GET', 'POST'])
     @auth_required()
     def register():
@@ -405,10 +269,11 @@ def create_application(config):
             else:
                 session['valid_thru'] = ''
             session['inactive'] = not form.active.data
-            return redirect('/store-token')
+            return redirect(f'/store-token')
         else:
             return render_template('register.html', token=door.get_most_recent_token(), form=form)
 
+
     @app.route('/edit-token/<token>', methods=['GET', 'POST'])
     @auth_required()
     def edit_token(token):
@@ -464,6 +329,7 @@ def create_application(config):
             else:
                 return render_template('edit.html', token=token, form=form)
 
+
     @app.route('/store-token')
     @auth_required()
     def store_token():
@@ -486,6 +352,7 @@ def create_application(config):
             flash(f"Error during store_tokens. Exception: {e}")
         return redirect('/tokens')
 
+
     @app.route('/delete-token/<token>', methods=['GET', 'POST'])
     @auth_required()
     def delete_token(token):
@@ -502,7 +369,7 @@ def create_application(config):
             token_to_delete = tokens[token]
 
             # set up form for confirming deletion
-            form = ConfirmDeleteForm()
+            form = TokenDeleteForm()
             form.name_confirm.data = token_to_delete['name']
 
             if request.method == 'GET':
@@ -520,13 +387,13 @@ def create_application(config):
                 return redirect('/tokens')
             else:
                 # form validation failed -> return to token overview and flash message
-                flash(
+                flash(f"Der eingegebene Name stimmt nicht überein. Der Token {token} von {token_to_delete['name']} wurde nicht gelöscht.")
-                    f"Der eingegebene Name stimmt nicht überein. Der Token {token} von {token_to_delete['name']} wurde nicht gelöscht.")
                 return redirect('/tokens')
         else:
             flash(f'Ungültiger Token {token} für Löschung.')
             return redirect('/tokens')
 
+
     @app.route('/deactivate-token/<token>')
     @auth_required()
     def deactivate_token(token):
@@ -547,19 +414,6 @@ def create_application(config):
                 flash(f"Error during store_tokens. Exception: {e}")
         return redirect('/tokens')
 
-    @app.route('/backup_tokens')
-    @auth_required()
-    def backup_tokens():
-        # get list of defined admin users for backup
-        tokens = door.get_tokens()
-        try:
-            with tempfile.TemporaryDirectory() as tmpdir:
-                file = Path(tmpdir, 'token_data.txt')
-                file.write_text(json.dumps(tokens))
-                return send_file(file, as_attachment=True, cache_timeout=-1)
-        except Exception as e:
-            return str(e)
-
     @app.route('/open')
     @auth_required()
     def open_door():
@@ -570,7 +424,6 @@ def create_application(config):
             flash(f'Could not open door. Exception: {e}')
         return redirect('/')
 
-    # routes for opening and closing the door via the web interface
     @app.route('/close')
     @auth_required()
     def close_door():
@@ -581,40 +434,4 @@ def create_application(config):
             flash(f'Could not close door. Exception: {e}')
         return redirect('/')
 
-    # setup user database when starting the app
-    with app.app_context():
-        new_admin_data = []
-        if config.admin_file is not None:
-            if not Path(config.admin_file).exists():
-                logger.warning(f"Admin user creation file not found at {config.admin_file}")
-            else:
-                # store data for new admins in memory s.t. the file can be deleted afterwards
-                with open(config.admin_file) as f:
-                    for i, d in enumerate(f.readlines()):
-                        try:
-                            user, email, pw = d.split()
-                            validate_email(email)
-                            new_admin_data.append({'username': user, 'email': email, 'password': pw})
-                        except Exception as e:
-                            print(
-                                f"Error while parsing line {i} in admin config file. Config file should contain lines of "
-                                f"'<username> <email> <password>\\n'\n Exception: {e}\nAdmin account could not be created.")
-
-        # create admin users (only if they don't exists already)
-        def create_super_admins(new_admin_data):
-            db.create_all()
-            super_admin_role = user_datastore.find_or_create_role('super_admin')  # root admin = can create other admins
-            admin_role = user_datastore.find_or_create_role('admin')  # 'normal' admin
-
-            for d in new_admin_data:
-                if user_datastore.find_user(email=d['email'], username=d['username']) is None:
-                    logger.info(f"New admin user created with username '{d['username']}' and email '{d['email']}'")
-                    # create new admin (only if admin does not already exist)
-                    new_admin = user_datastore.create_user(email=d['email'], username=d['username'],
-                                                           password=hash_password(d['password']),
-                                                           roles=[super_admin_role, admin_role])
-            db.session.commit()
-
-        create_super_admins(new_admin_data)
-
     return app