import os from flask import Flask, render_template, request, flash, redirect, session from flask_wtf import FlaskForm from wtforms.fields.html5 import DateField, EmailField from wtforms.fields import StringField, BooleanField from wtforms.validators import DataRequired, ValidationError from flask_sqlalchemy import SQLAlchemy from flask_security import Security, SQLAlchemyUserDatastore, auth_required, hash_password from flask_security.models import fsqla_v2 as fsqla from flask_security.forms import LoginForm, Required, PasswordField from datetime import date from door import Door door = Door('10.10.21.2') app = Flask(__name__) # Generate a nice key using secrets.token_urlsafe() app.config['SECRET_KEY'] = os.environ.get("SECRET_KEY", 'Q7PJu2fg2jabYwP-Psop6c6f2G4') # Bcrypt is set as default SECURITY_PASSWORD_HASH, which requires a salt # Generate a good salt using: secrets.SystemRandom().getrandbits(128) app.config['SECURITY_PASSWORD_SALT'] = os.environ.get("SECURITY_PASSWORD_SALT", '10036796768252925167749545152988277953') app.config['SECURITY_USER_IDENTITY_ATTRIBUTES'] = ('username', 'email') app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite://' # As of Flask-SQLAlchemy 2.4.0 it is easy to pass in options directly to the # underlying engine. This option makes sure that DB connections from the # pool are still valid. Important for entire application since # many DBaaS options automatically close idle connections. app.config["SQLALCHEMY_ENGINE_OPTIONS"] = { "pool_pre_ping": True, } # Create database connection object db = SQLAlchemy(app) # Define models fsqla.FsModels.set_db_info(db) class Role(db.Model, fsqla.FsRoleMixin): pass class User(db.Model, fsqla.FsUserMixin): username = db.Column(db.String(255)) class ExtendedLoginForm(LoginForm): email = StringField('Benutzername oder E-Mail', [Required()]) password = PasswordField('Passwort', [Required()]) # Setup Flask-Security user_datastore = SQLAlchemyUserDatastore(db, User, Role) security = Security(app, user_datastore, login_form=ExtendedLoginForm) def validate_valid_thru_date(form, field): if form.limit_validity.data: # only check date format if limited validity of token is set try: if not field.data >= date.today(): raise ValueError except ValueError as e: flash("Ungültiges Datum") raise ValidationError return True class TokenForm(FlaskForm): name = StringField('Name', validators=[DataRequired()]) email = EmailField('E-Mail', validators=[DataRequired()]) limit_validity = BooleanField('Gültigkeit begrenzen?') valid_thru = DateField('Gültig bis', validators=[validate_valid_thru_date]) active = BooleanField('Aktiv?') dsgvo = BooleanField('Einwilligung DSGVO erfragt?', validators=[DataRequired()]) # Create a user to test with @app.before_first_request def create_user(): db.create_all() if not user_datastore.find_user(email='admin@example.com', username="admin"): user_datastore.create_user(email='admin@example.com', username="admin", password=hash_password("password")) db.session.commit() @app.route('/') def door_lock(): return render_template('index.html', door_state=door.state, encoder_position=door.encoder_position) @app.route('/tokens') #@auth_required() def list_tokens(): tokens = door.get_tokens() assigned_tokens = {t: data for t, data in tokens.items() if not data['inactive']} 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) @app.route('/register-token', methods=['GET', 'POST']) #@auth_required() def register(): """Register new token for locking and unlocking the door. This route displays the most recently scanned invalid token as reported in the logfile and provides a form for entering user info (name, email, valid thru date (optional)) for the new token. If the route is called via POST the provided form data is checked and if the check succeeds the /store-token route will be called which adds the new token to the database. """ form = TokenForm() if request.method == 'GET': # set default valid thru date to today to make sure form validity check passes # (will not be used if limited validity is disabled) form.valid_thru.data = date.today() return render_template('register.html', token=door.get_most_recent_token(), form=form) elif request.method == 'POST' and form.validate(): # store data in session cookie session['token'] = door.get_most_recent_token()['token'] session['name'] = form.name.data session['email'] = form.email.data if form.limit_validity.data: session['valid_thru'] = form.valid_thru.data.isoformat() else: session['valid_thru'] = '' session['inactive'] = not form.active.data return redirect(f'/store-token') else: return render_template('register.html', token=door.get_most_recent_token(), form=form) @app.route('/edit-token/', methods=['GET', 'POST']) #@auth_required() def edit_token(token): """Edit data in the token file (name, email, valid_thru date, active/inactive). If the route is accessed via GET it will provide a form for editing the currently stored data for the user. If the route is accessed via POST it will check if the form data is good and then store the modified user data in the database (by redirecting to the /store-token route) Parameters ---------- token : str The token for which data should be edited. """ form = TokenForm() form.dsgvo.validators = [] # we skip the validation of the DSGVO checkbox here because we assume the user agreed # to it before if request.method == 'GET': tokens = door.get_tokens() if token in tokens: # set default for form according to values from the token file et = tokens[token] form.active.data = not et['inactive'] form.name.data = et['name'] if et['name'] else '' form.email.data = et['email'] if et['email'] else '' # for the valid thru date we use today's date in case there is not valid date in the database try: form.valid_thru.data = date.fromisoformat(et['valid_thru']) form.limit_validity.data = True except Exception: form.valid_thru.data = date.today() return render_template('edit.html',token=token, form=form) else: # flash an error message if the route is accessed with an invalid token flash(f'Ausgewaehlter Token {token} in Tokenfile nicht gefunden.') return redirect('/tokens') elif request.method == 'POST' and form.validate(): # store data in session cookie session['token'] = token session['name'] = form.name.data session['email'] = form.email.data if form.limit_validity.data: session['valid_thru'] = form.valid_thru.data.isoformat() else: session['valid_thru'] = '' session['inactive'] = not form.active.data return redirect(f'/store-token') @app.route('/store-token') #@auth_required() def store_token(): """Store token to the token file on disk. This will use the token id and the associated data stored in the session cookie (filled by register_token() or edit_token()) and create/modify a token and store the new token file to disk. """ token = session['token'] tokens = door.get_tokens() tokens[token] = {'name': session['name'], 'email': session['email'], 'valid_thru': session['valid_thru'], 'inactive': session['inactive'], 'organization': 'test_org'} door.store_tokens(tokens) return redirect('/tokens') @app.route('/delete-token', methods=['POST']) #@auth_required() def delete_token(): """Delete the given token from the token file and store the new token file to disk Parameters ---------- token : str The token to delete from the database. """ token = request.form.get('token') tokens = door.get_tokens() if token in tokens: # check if token exists tokens.pop(token) door.store_tokens(tokens) return "success" @app.route('/deactivate-token/') #@auth_required() def deactivate_token(token): """Deactivate access for the given token. This updates the token file on disk. Parameters ---------- token : str The token to deactivate. """ tokens = door.get_tokens() if token in tokens: tokens[token]['inactive'] = True door.store_tokens(tokens) return redirect('/tokens') if __name__ == '__main__': app.run()