guille
aa76eff3ab
Correct swagger definition. MEB uses body, so should be a post method not a get according to REST definition. active_session now refresh the session also if valid. partial implementation for meb.
78 lines
2.1 KiB
Python
78 lines
2.1 KiB
Python
"""
|
|
Session helper
|
|
SPDX - License - Identifier: LGPL - 3.0 - or -later
|
|
Copyright © 2022 Project Author name guillermo.gutierrezmorote@concordia.ca
|
|
"""
|
|
import uuid
|
|
import datetime
|
|
import time
|
|
|
|
sessions = {}
|
|
begin_time = None
|
|
swagger_data = None
|
|
city = None
|
|
greenery_catalog = None
|
|
construction_catalog = None
|
|
usage_catalog = None
|
|
debug_mode = False
|
|
|
|
def expired_sessions_collector(session_timeout_duration):
|
|
"""
|
|
Goes through each session in sessions and removes expired session(s)
|
|
"""
|
|
while True:
|
|
if bool(sessions):
|
|
for session in list(sessions):
|
|
_expire = datetime.datetime.strptime(sessions[session]['expire'], '%Y-%m-%d %H:%M:%S.%f')
|
|
if _expire < datetime.datetime.now():
|
|
print("session with session_id: ", session, "expired.")
|
|
del sessions[session]
|
|
time.sleep(60 * int(session_timeout_duration))
|
|
|
|
|
|
def _validate_session(session_id, token, application_uuid):
|
|
"""
|
|
Checks if session is valid
|
|
"""
|
|
try:
|
|
session = sessions[session_id]
|
|
if debug_mode:
|
|
token = session['token']
|
|
return bool(session) and (session['token'] == token) and session['application_uuid'] == application_uuid
|
|
except KeyError:
|
|
return False
|
|
|
|
|
|
def remove_session(session_id, token, application_uuid):
|
|
"""
|
|
Remove a session from the sessions array
|
|
"""
|
|
if _validate_session(session_id, token, application_uuid):
|
|
del sessions[session_id]
|
|
return True
|
|
return False
|
|
|
|
|
|
def refresh_session(session_id, token, application_uuid):
|
|
"""
|
|
Validate and extend current session
|
|
:return: valid session
|
|
"""
|
|
if _validate_session(session_id, token, application_uuid):
|
|
sessions[session_id]['expire'] = str(datetime.datetime.now() + datetime.timedelta(minutes=5))
|
|
sessions[session_id]['token'] = str(uuid.uuid4())
|
|
return sessions[session_id]
|
|
|
|
return None
|
|
|
|
|
|
def active_session(session_id, token, application_uuid):
|
|
_is_valid = _validate_session(session_id=session_id, token=token, application_uuid=application_uuid)
|
|
if _is_valid:
|
|
refresh_session(session_id, token, application_uuid)
|
|
return _is_valid
|
|
|
|
|
|
def session(session_id) -> {}:
|
|
return sessions[session_id]
|