DoorControl/door_pi_control/door/token_control.py

90 lines
3.2 KiB
Python
Raw Normal View History

2022-11-06 13:30:11 +00:00
from ... import util
import datetime
import os
import select
from threading import Thread, RLock
class TokenControl(util.Loggable):
def __init__(self, token_path, fifo_path, control):
self._token_path = token_path
self._fifo_path = fifo_path
self._control = control
self._open_fifo()
self.refresh_tokens()
pipes = os.pipe()
self._mutex = RLock()
self._stop_pipe_write = pipes[1]
self._stop_pipe_read = pipes[0]
self._thread = None
def _open_fifo(self):
self._fifo = open(self._fifo_path, "r")
def refresh_tokens(self):
"""Refreshes all tokens from config.valid_tokens"""
valid = {}
try:
self._logger().info("Loading tokens")
lines =[ s.strip() for s in open(self._token_path, "r").readlines() ]
for i, line in enumerate(lines):
l = line.split('|')
if len(l) == 5:
if not l[0].strip().startswith('#'):
token, name, organization, email, valid_thru = l
try:
if len(valid_thru.strip()) > 0:
valid_thru = datetime.date.fromisoformat(valid_thru)
else:
valid_thru = None
except Exception:
self._logger().error(f"Could not parse valid thru date for token {token} in line {i}")
valid_thru = None
self._logger().debug(f"Got token {token} associated with {name} <{email}> of {organization}, valid thru {valid_thru}")
if token in valid:
self._logger().warning(f"Overwriting token {token}")
valid[token] = {
'name': name,
'organization': organization,
'email': email,
'valid_thru': valid_thru
}
else:
self._logger().warning(f"Skipping line {i} ({line}) since it does not contain exactly 5 data field")
except Exception as e:
valid = {}
self._logger().error(f"Error reading token file. Exception: {e}")
with self._mutex:
self._tokens = valid
def start(self):
with self._mutex:
if self._thread != None:
return
self._thread = Thread(target = self.run, daemon=True)
self._thread.start()
def stop(self):
with self._mutex:
if self._thread == None:
return
self._stop_pipe_write.write('c')
self._thread.join()
self._thread = None
def run(self):
while True:
readable = select.select([self._fifo, self._stop_pipe_read], [], [], 0.5)[0]
if self._stop_pipe_read in readable:
self._stop_pipe_read.read(1)
return
if self._fifo in readable:
line = self._fifo.readline()
if line == None:
self._open_fifo()
continue