130 lines
3.2 KiB
Python
130 lines
3.2 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
|
|
|
|
sessions = {}
|
|
begin_time = None
|
|
swagger_data = None
|
|
city = None
|
|
greenery_catalog = None
|
|
construction_catalog = None
|
|
usage_catalog = None
|
|
|
|
class SessionData:
|
|
def __init__(self, session):
|
|
self._session = session
|
|
|
|
@property
|
|
def city(self):
|
|
return self._session['city']
|
|
|
|
@property
|
|
def id(self):
|
|
return self._session['session_id']
|
|
|
|
@property
|
|
def token(self):
|
|
return self._session['token']
|
|
|
|
@property
|
|
def headers(self):
|
|
return {'session_id': str(self.id), 'token': str(self.token)}
|
|
|
|
@property
|
|
def greenery_percentage(self):
|
|
return self._session['greenery_percentage']
|
|
|
|
@greenery_percentage.setter
|
|
def greenery_percentage(self, value):
|
|
self._session['greenery_percentage'] = value
|
|
|
|
@property
|
|
def greenery_catalog(self):
|
|
return greenery_catalog
|
|
|
|
@property
|
|
def construction_catalog(self):
|
|
return construction_catalog
|
|
|
|
@property
|
|
def usage_catalog(self):
|
|
return usage_catalog
|
|
|
|
|
|
def clear_old_sessions():
|
|
#loop through all sessions and remove expired sessions
|
|
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():
|
|
del sessions[session]
|
|
|
|
|
|
def _validate_session(session_id, token, application_id):
|
|
print(sessions)
|
|
print(session_id)
|
|
print(token)
|
|
print(application_id)
|
|
if bool(sessions[session_id]) and sessions[session_id]['token'] == token and \
|
|
sessions[session_id]['application_id'] == application_id:
|
|
return True
|
|
return False
|
|
|
|
|
|
def remove_session(request):
|
|
"""
|
|
Remove a session from the sessions array
|
|
"""
|
|
session = eval(request.headers.get('session'))
|
|
session_id = session['session_id']
|
|
token = session['token']
|
|
ip = request.remote_addr
|
|
valid, i = _valid_session(session_id, ip, token)
|
|
if valid:
|
|
del sessions[i]
|
|
return valid
|
|
|
|
|
|
def refresh_session(request):
|
|
"""
|
|
Validate and extend current session
|
|
:return: valid, token, city
|
|
"""
|
|
session_id = request.headers.get('session_id', None)
|
|
token = request.headers.get('token', None)
|
|
application_id = request.headers.get('application_id', None)
|
|
|
|
if _validate_session(session_id, token, application_id):
|
|
sessions[session_id]['expire'] = str(datetime.datetime.now() + datetime.timedelta(minutes=5))
|
|
sessions[session_id]['token'] = uuid.uuid4()
|
|
return sessions[session_id]
|
|
|
|
return None
|
|
'''
|
|
session_header = request.headers.get('session')
|
|
print(session_header)
|
|
if session_header is None:
|
|
return None
|
|
session = eval(session_header)
|
|
session_id = session['session_id']
|
|
token = session['token']
|
|
ip = request.remote_addr
|
|
valid, i = _valid_session(session_id, ip, token)
|
|
if valid:
|
|
if session_id == 'debug':
|
|
new_token = 'debug'
|
|
else:
|
|
new_token = str(uuid.uuid4())
|
|
expire = str(datetime.datetime.now() + datetime.timedelta(minutes=5))
|
|
sessions[i]['token'] = new_token
|
|
sessions[i]['expire'] = expire
|
|
return SessionData(sessions[i])
|
|
else:
|
|
return None
|
|
'''
|