1
0
mirror of https://github.com/binary-kitchen/doorlockd synced 2024-10-31 22:47:05 +01:00
doorlockd-mirror/doorlockd
Ralf Ramsauer badeb1945b pydoorlock: Authenticator: Move all auth-related logic to Authenticator
By passing the configuration

Signed-off-by: Ralf Ramsauer <ralf@binary-kitchen.de>
2018-10-16 21:25:10 +02:00

427 lines
13 KiB
Python
Executable File

#!/usr/bin/env python3
"""
Doorlockd -- Binary Kitchen's smart door opener
Copyright (c) Binary Kitchen e.V., 2018
Author:
Ralf Ramsauer <ralf@binary-kitchen.de>
This work is licensed under the terms of the GNU GPL, version 2. See
the LICENSE file in the top-level directory.
This program is distributed in the hope that it will be useful, but WITHOUT
ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
details.
"""
import logging
import sys
from configparser import ConfigParser
from enum import Enum
from os.path import join
from serial import Serial
from subprocess import Popen
from threading import Thread
from time import sleep
from flask import abort, Flask, jsonify, render_template, request
from flask_socketio import SocketIO
from flask_wtf import FlaskForm
from wtforms import PasswordField, StringField, SubmitField
from wtforms.validators import DataRequired, Length
from pydoorlock.Authenticator import Authenticator, AuthMethod, AuthenticationResult
from pydoorlock.WebApp import AuthenticationForm
from pydoorlock.Door import DoorState
SYSCONFDIR = '.'
PREFIX = '.'
root_prefix = join(PREFIX, 'share', 'doorlockd')
sounds_prefix = join(root_prefix, 'sounds')
static_folder = join(root_prefix, 'static')
template_folder = join(root_prefix, 'templates')
scripts_prefix = join(root_prefix, 'scripts')
__author__ = 'Ralf Ramsauer'
__copyright = 'Copyright (c) Ralf Ramsauer, 2018'
__license__ = 'GPLv2'
__email__ = 'ralf@binary-kitchen.de'
__status__ = 'Development'
__maintainer__ = 'Ralf Ramsauer'
__version__ = '2.0-rc2'
log_level = logging.DEBUG
date_fmt = '%Y-%m-%d %H:%M:%S'
log_fmt = '%(asctime)-15s %(levelname)-8s %(message)s'
log = logging.getLogger()
webapp = Flask(__name__,
template_folder=template_folder,
static_folder=static_folder)
socketio = SocketIO(webapp, async_mode='threading')
class Config:
config_topic = 'doorlock'
def __init__(self, sysconfdir):
self.config = ConfigParser()
self.config.read([join(sysconfdir, 'doorlockd.default.cfg'),
join(sysconfdir, 'doorlockd.cfg')])
def boolean(self, key):
return self.config.getboolean(self.config_topic, key)
def str(self, key):
return self.config.get(self.config_topic, key)
cfg = Config(SYSCONFDIR)
# Booleans
debug = cfg.boolean('DEBUG')
simulate_serial = cfg.boolean('SIMULATE_SERIAL')
run_hooks = cfg.boolean('RUN_HOOKS')
sounds = cfg.boolean('SOUNDS')
serial_port = cfg.str('SERIAL_PORT')
room = cfg.str('ROOM')
title = cfg.str('TITLE')
welcome = cfg.str('WELCOME')
webapp.config['SECRET_KEY'] = cfg.str('SECRET_KEY')
html_title = '%s (%s - v%s)' % (title, __status__, __version__)
wave_emergency = 'emergency_unlock.wav'
wave_lock = 'lock.wav'
wave_lock_button = 'lock_button.wav'
wave_present = 'present.wav'
wave_present_button = 'present.wav'
wave_unlock = 'unlock.wav'
wave_unlock_button = 'unlock_button.wav'
wave_zonk = 'zonk.wav'
host = 'localhost'
if debug:
host = '0.0.0.0'
def playsound(filename):
if not sounds:
return
Popen(['nohup', 'aplay', join(sounds_prefix, filename)])
def start_hook(script):
if not run_hooks:
log.info('Hooks disabled: not starting %s' % script)
return
log.info('Starting hook %s' % script)
Popen(['nohup', join(scripts_prefix, script)])
def sound_helper(old_state, new_state, button):
if old_state == new_state:
playsound(wave_zonk)
if button:
if new_state == DoorState.Open:
playsound(wave_unlock_button)
elif new_state == DoorState.Present:
playsound(wave_present_button)
elif new_state == DoorState.Closed:
playsound(wave_lock_button)
else:
if new_state == DoorState.Open:
playsound(wave_unlock)
elif new_state == DoorState.Present:
playsound(wave_present)
elif new_state == DoorState.Closed:
playsound(wave_lock)
class LogicResponse(Enum):
Success = 0
Perm = 1
AlreadyActive = 2
# don't break old apps, value 3 is reserved now
RESERVED = 3
Inval = 4
EmergencyUnlock = 10,
ButtonLock = 11,
ButtonUnlock = 12,
ButtonPresent = 13,
def __str__(self):
if self == LogicResponse.Success:
return 'Yo, passt.'
elif self == LogicResponse.Perm:
return choose_insult()
elif self == LogicResponse.AlreadyActive:
return 'Zustand bereits aktiv'
elif self == LogicResponse.Inval:
return 'Das was du willst geht nicht.'
elif self == LogicResponse.EmergencyUnlock:
return '!!! Emergency Unlock !!!'
elif self == LogicResponse.ButtonLock:
return 'Closed by button'
elif self == LogicResponse.ButtonUnlock:
return 'Opened by button'
elif self == LogicResponse.ButtonPresent:
return 'Present by button'
return 'Error'
class DoorHandler:
state = DoorState.Closed
do_close = False
CMD_PRESENT = b'y'
CMD_OPEN = b'g'
CMD_CLOSE = b'r'
BUTTON_PRESENT = b'Y'
BUTTON_OPEN = b'G'
BUTTON_CLOSE = b'R'
CMD_EMERGENCY_SWITCH = b'E'
# TBD DOOR NOT CLOSED
def __init__(self, device):
if simulate_serial:
return
self.serial = Serial(device, baudrate=9600, bytesize=8, parity='N',
stopbits=1, timeout=0)
self.thread = Thread(target=self.thread_worker)
log.debug('Spawning RS232 Thread')
self.thread.start()
def thread_worker(self):
while True:
sleep(0.4)
while True:
rx = self.serial.read(1)
if len(rx) == 0:
break
old_state = self.state
if rx == DoorHandler.BUTTON_CLOSE:
self.close()
log.info('Closed due to Button press')
logic.emit_status(LogicResponse.ButtonLock)
elif rx == DoorHandler.BUTTON_OPEN:
self.open()
log.info('Opened due to Button press')
logic.emit_status(LogicResponse.ButtonUnlock)
elif rx == DoorHandler.BUTTON_PRESENT:
self.present()
log.info('Present due to Button press')
logic.emit_status(LogicResponse.ButtonPresent)
elif rx == DoorHandler.CMD_EMERGENCY_SWITCH:
log.warning('Emergency unlock')
logic.emit_status(LogicResponse.EmergencyUnlock)
else:
log.error('Received unknown message "%s" from AVR' % rx)
sound_helper(old_state, self.state, True)
if self.do_close:
tx = DoorHandler.CMD_CLOSE
self.do_close = False
elif self.state == DoorState.Present:
tx = DoorHandler.CMD_PRESENT
elif self.state == DoorState.Open:
tx = DoorHandler.CMD_OPEN
else:
continue
self.serial.write(tx)
self.serial.flush()
def open(self):
if self.state == DoorState.Open:
return LogicResponse.AlreadyActive
self.state = DoorState.Open
start_hook('post_unlock')
return LogicResponse.Success
def present(self):
if self.state == DoorState.Present:
return LogicResponse.AlreadyActive
self.state = DoorState.Present
start_hook('post_present')
return LogicResponse.Success
def close(self):
if self.state == DoorState.Closed:
return LogicResponse.AlreadyActive
self.do_close = True
self.state = DoorState.Closed
start_hook('post_lock')
return LogicResponse.Success
def request(self, state):
if state == DoorState.Closed:
return self.close()
elif state == DoorState.Present:
return self.present()
elif state == DoorState.Open:
return self.open()
class Logic:
def __init__(self):
self.auth = Authenticator(cfg)
self.door_handler = DoorHandler(serial_port)
def _request(self, state, credentials):
err = self.auth.try_auth(credentials)
if err != AuthenticationResult.Success:
return err
return self.door_handler.request(state)
def request(self, state, credentials):
old_state = self.door_handler.state
err = self._request(state, credentials)
if err == LogicResponse.Success or err == LogicResponse.AlreadyActive:
sound_helper(old_state, self.door_handler.state, False)
self.emit_status(err)
return err
def emit_status(self, message=None):
led = self.state.to_img()
if message is None:
message = self.state.to_html()
else:
message = str(message)
socketio.emit('status', {'led': led, 'message': message})
@property
def state(self):
return self.door_handler.state
@socketio.on('request_status')
@socketio.on('connect')
def on_connect():
logic.emit_status()
@webapp.route('/display')
def display():
return render_template('display.html',
room=room,
title=title,
welcome=welcome)
@webapp.route('/api', methods=['POST'])
def api():
def json_response(response, msg=None):
json = dict()
json['err'] = response.value
json['msg'] = response.to_html() if msg is None else msg
if response == LogicResponse.Success or \
response == LogicResponse.AlreadyActive:
# TBD: Remove 'open'. No more users. Still used in App Version 2.1.1!
json['open'] = logic.state.is_open()
json['status'] = logic.state.value
return jsonify(json)
method = request.form.get('method')
user = request.form.get('user')
password = request.form.get('pass')
command = request.form.get('command')
if method == 'local':
method = AuthMethod.LOCAL_USER_DB
else: # 'ldap' or default
method = AuthMethod.LDAP_USER_PW
if any(v is None for v in [user, password, command]):
log.warning('Incomplete API request')
abort(400)
if not user or not password:
log.warning('Invalid username or password format')
return json_response(LogicResponse.Inval,
'Invalid username or password format')
credentials = method, user, password
if command == 'status':
return json_response(logic.try_auth(credentials))
desired_state = DoorState.from_string(command)
if not desired_state:
return json_response(LogicResponse.Inval, "Invalid command requested")
log.info('Incoming API request from %s' % user.encode('utf-8'))
log.info(' desired state: %s' % desired_state)
log.info(' current state: %s' % logic.state)
response = logic.request(desired_state, credentials)
return json_response(response)
@webapp.route('/', methods=['GET', 'POST'])
def home():
authentication_form = AuthenticationForm()
response = None
if request.method == 'POST' and authentication_form.validate():
user = authentication_form.username.data
password = authentication_form.password.data
method = authentication_form.method
credentials = method, user, password
log.info('Incoming request from %s' % user.encode('utf-8'))
log.info(' authentication method: %s' % method)
desired_state = authentication_form.desired_state
log.info(' desired state: %s' % desired_state)
log.info(' current state: %s' % logic.state)
response = logic.request(desired_state, credentials)
log.info(' response: %s' % response)
# Don't trust python, zero credentials
user = password = credentials = None
return render_template('index.html',
authentication_form=authentication_form,
auth_backends=logic.auth.backends,
response=response,
state_text=logic.state.to_html(),
led=logic.state.to_img(),
banner='%s - %s' % (title, room))
if __name__ == '__main__':
logging.basicConfig(level=log_level, stream=sys.stdout,
format=log_fmt, datefmt=date_fmt)
log.info('Starting doorlockd')
log.info('Using serial port: %s' % serial_port)
logic = Logic()
socketio.run(webapp, host=host, port=8080, use_reloader=False, debug=debug)
sys.exit(0)