Telos4/DoorAdmin
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

improved admin user creation and added basic logging features

Browse Source
This commit is contained in:
Simon Pirkelmann 2021-03-27 16:41:51 +01:00
parent eb4c027f46
commit 4197446a00
4 changed files with 87 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
bin/launch_webadmin
View File

@ -9,6 +9,7 @@ parser.add_argument("--nfc_socket", default="/tmp/nfc.sock", help="socket for ha
parser.add_argument("--template_folder", default="templates", help="path to Flask templates folder")
parser.add_argument("--static_folder", default="static", help="path to Flask static folder")
parser.add_argument("--admin_file", help="Path to file for creating initial admin users")
parser.add_argument("--log_file", default="/var/log/webinterface.log", help="Path to log file")
parser.add_argument("--ldap_url", default="ldaps://do.imaginaerraum.de",
help="URL for LDAP server for alternative user authorization")
parser.add_argument("--mqtt_host", default="10.10.21.2", help="IP address of MQTT broker")

21
imaginaerraum_door_admin/door_handle.py
View File

@ -1,10 +1,10 @@
import paho.mqtt.client as mqtt
import socket
from pathlib import Path
import logging
class DoorHandle:
def __init__(self, token_file, mqtt_host, mqtt_port=1883, nfc_socket='/tmp/nfc.sock'):
def __init__(self, token_file, mqtt_host, mqtt_port=1883, nfc_socket='/tmp/nfc.sock', logger=None):
self.state = None
self.encoder_position = None
@ -20,18 +20,25 @@ class DoorHandle:
self.mqtt_client.connect_async(host=mqtt_host, port=mqtt_port)
self.mqtt_client.loop_start()
if logger:
self.logger = logger
else:
self.logger = logging
self.nfc_sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
try:
self.nfc_sock.connect(nfc_socket)
self.logger.info(f"Connected to NFC socket at {nfc_socket}.")
except Exception as e:
print(f"Could not connect to NFC socket at {nfc_socket}. Exception: {e}")
self.nfc_sock = None
#raise
self.data_fields = ['name', 'organization', 'email', 'valid_thru']
# The callback for when the client receives a CONNACK response from the server.
def on_connect(self, client, userdata, flags, rc):
print("Connected with result code " + str(rc))
self.logger.info("Connected to MQTT broker with result code " + str(rc))
# Subscribing in on_connect() means that if we lose the connection and
# reconnect then subscriptions will be renewed.
@ -81,13 +88,21 @@ class DoorHandle:
# write new tokens to file and trigger reload
with open(self.token_file, 'w') as f:
f.write(output)
if self.nfc_sock is not None:
self.nfc_sock.send(b'rld\n')
def open_door(self):
if self.nfc_sock is not None:
self.nfc_sock.send(b'open\n')
else:
raise Exception("No connection to NFC socket. Cannot close door!")
def close_door(self):
if self.nfc_sock is not None:
self.nfc_sock.send(b'close\n')
else:
raise Exception("No connection to NFC socket. Cannot close door!")
def get_most_recent_token(self):
# read last invalid token from logfile

2
imaginaerraum_door_admin/templates/tokens.html
View File

@ -34,7 +34,7 @@
{% endfor %}
<td>
<a href="{{ url_for('edit_token', token=t) }}"><img src="static/edit.png" title="Editieren" alt="Edit"></a>
<img src="static/delete.png" title="Löschen" alt="Delete" onclick="confirmDelete('{{ t }}')">
<a href="{{ url_for('delete_token', token=t) }}"><img src="static/delete.png" title="Löschen" alt="Delete"></a>
</td>
</tr>
{% endfor %}

80
imaginaerraum_door_admin/webapp.py
View File

@ -5,7 +5,7 @@ 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
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
@ -14,6 +14,8 @@ from email_validator import validate_email
import bleach
import ldap3
from ldap3.core.exceptions import LDAPBindError, LDAPSocketOpenError
from pathlib import Path
import logging
from datetime import date
from .door_handle import DoorHandle
@ -48,8 +50,51 @@ def uia_username_mapper(identity):
return bleach.clean(identity, strip=True)
def create_application(config):
# set up logging for the web app
logger = logging.getLogger('webapp')
logger.setLevel(logging.INFO)
if config.log_file is not None:
ch = logging.FileHandler(config.log_file)
ch.setLevel(logging.INFO)
else:
# create console handler and set level to debug
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
# create formatter
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
# add formatter to ch
ch.setFormatter(formatter)
# add ch to logger
logger.addHandler(ch)
# do some checks for file existence etc.
if not Path(config.template_folder).exists():
logger.error(f'Flask template folder not found at {config.template_folder}')
if not Path(config.static_folder).exists():
logger.error(f'Flask static folder not found at {config.static_folder}')
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)
door = DoorHandle(token_file=config.token_file, mqtt_host=config.mqtt_host, nfc_socket=config.nfc_socket,
logger=logger)
app = Flask(__name__, template_folder=config.template_folder, static_folder=config.static_folder)
@ -153,6 +198,10 @@ def create_application(config):
# 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
return authorized
@ -162,28 +211,20 @@ def create_application(config):
security = Security(app, user_datastore, login_form=ExtendedLoginForm)
# create admin users (only if they don't exists already)
def create_admins(admin_user_file):
with open(admin_user_file) as f:
admin_data = f.readlines()
for i, d in enumerate(admin_data):
try:
user, email, pw = d.split()
if user_datastore.find_user(email=email, username=user) is None:
validate_email(email)
def create_admins(new_admin_data):
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)
user_datastore.create_user(email=email, username=user, password=hash_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.")
db.session.commit()
user_datastore.create_user(email=d['email'], username=d['username'], password=hash_password(d['password']))
# Create a user to test with
@app.before_first_request
def create_user():
db.create_all()
if config.admin_file is not None:
if len(new_admin_data) > 0:
# create admin accounts from given file
create_admins(config.admin_file)
create_admins(new_admin_data)
db.session.commit()
@app.route('/')
@ -306,6 +347,7 @@ def create_application(config):
'organization': session['organization']}
try:
door.store_tokens(tokens)
logger.info(f"Token {token} stored in database by admin user {current_user.username}")
except Exception as e:
flash(f"Error during store_tokens. Exception: {e}")
return redirect('/tokens')
@ -338,6 +380,7 @@ def create_application(config):
tokens.pop(token)
try:
door.store_tokens(tokens)
logger.info(f"Token {token} was deleted from database by admin user {current_user.username}")
except Exception as e:
flash(f"Error during store_tokens. Exception: {e}")
flash(f"Token {token} wurde gelöscht!")
@ -366,6 +409,7 @@ def create_application(config):
tokens[token]['inactive'] = True
try:
door.store_tokens(tokens)
logger.info(f"Token {token} deactivated by admin user {current_user.username}")
except Exception as e:
flash(f"Error during store_tokens. Exception: {e}")
return redirect('/tokens')
@ -375,6 +419,7 @@ def create_application(config):
def open_door():
try:
door.open_door()
logger.info(f"Door opened by admin user {current_user.username}")
except Exception as e:
flash(f'Could not open door. Exception: {e}')
return redirect('/')
@ -384,6 +429,7 @@ def create_application(config):
def close_door():
try:
door.close_door()
logger.info(f"Door closed by admin user {current_user.username}")
except Exception as e:
flash(f'Could not close door. Exception: {e}')
return redirect('/')