Elyos_FI_Back_Office/user_session.py

143 lines
4.9 KiB
Python

'''
Ce fichier est tout ce qui est lié à la session de connexion
- check login/pass
- creation et gestion token de session
- etc.
'''
import secrets
from pymongo import MongoClient
import json
from bson import ObjectId
from datetime import datetime
import logging
import prj_common as mycommon
from pymongo import ReturnDocument
import inspect
import sys, os
TOKEN_SIZE = 25
CONNECTION_STRING = "mongodb://localhost/cherifdb"
client = MongoClient(CONNECTION_STRING)
dbname = client['cherifdb']
class JSONEncoder(json.JSONEncoder):
def default(self, o):
if isinstance(o, ObjectId):
return str(o)
return json.JSONEncoder.default(self, o)
def create_token_urlsafe():
return secrets.token_urlsafe(TOKEN_SIZE)
'''
Lorsqu'un user se logue, on recupere son RecId,
puis on enregiste dans table des token :
- le recid,
- le token
'''
def login(diction):
try:
# Dictionnaire des champs utilisables
field_list = ['email', 'pwd']
incom_keys = diction.keys()
'''
# Verification que les champs reçus dans l'API sont bien dans la liste des champs autorisés
# Cela evite le cas ou une entité tierce ajouter les valeurs inconnu dans l'API
# Ici on doit mettre tous les champs possible (obligatoire ou non) de la BDD dans la liste
# field_list.
'''
for val in incom_keys:
if str(val).lower() not in str(field_list).lower():
mycommon.myprint(str(inspect.stack()[0][3])+" Le champ '" + val + "' n'existe pas, MAJ annulée")
return False, " Impossible de se connecter, merci de contacter le support technique"
'''
Une fois qu'on a controlé que toutes les clés mise dans l'API sont correcte. etape precedente,
On controle que les champs obligatoires sont presents dans la liste
'''
field_list_obligatoire = ['email', 'pwd']
for val in field_list_obligatoire:
if str(val).lower() not in diction:
mycommon.myprint(str(inspect.stack()[0][3])+" La valeur '" + val + "' n'est pas presente dans liste ")
return False, " Impossible de se connecter, merci de contacter le support technique"
query = {}
# recuperation des paramettre
if ("email" in diction.keys()):
if diction['email']:
email = diction['email']
query['email'] = diction['email']
query['active'] = '1'
if ("pwd" in diction.keys()):
if diction['pwd']:
pwd = diction['pwd']
query['pwd'] = diction['pwd']
# Recuperation du RecId du user connecté
coll_name = dbname['user_account']
print(" on recherche ici " + email)
# Verification si ce user existe (controle : login et passwd)
tmp = coll_name.count_documents(query)
if (tmp != 1):
mycommon.myprint(str(inspect.stack()[0][3])+" Login/mot de passe invalide " + str(query)+ " my_user.count() = "+str(tmp))
return False, " Login/mot de passe invalide, merci de verifier votre saisie. "
my_user = coll_name.find(query)
#recuperation du recid du user
user_recid = my_user[0]['recid']
print("user_recid = "+user_recid)
# A ce niveau le login et passe sont ok.
# il faut donc créer le token et renvoyer le token.
my_token = create_token_urlsafe()
# Enregistrement du token
dbname2 = client['cherifdb']
token_collection = dbname2['user_token']
tokenfield = {}
tokenfield['token'] = my_token
tokenfield['email'] = email
tokenfield['recid'] = user_recid
tokenfield['date_update'] = str(datetime.now())
tokenfield['valide'] = '1'
'''
Si il y a un token en cours de validité, il est mis à jour pr prolonger la durée
de vie du token
s'il n'y a pas de token valide, aloes on insert un.
'''
ret_val = token_collection.find_one_and_update({'recid': str(user_recid), 'valide': '1'},
{"$set": tokenfield},
upsert=True,
return_document=ReturnDocument.AFTER
)
if ret_val and ret_val['_id']:
mycommon.myprint(" Ajout Token OK, _id = "+str(ret_val['_id']))
return True, str(my_token)
else:
mycommon.myprint(str(inspect.stack()[0][3])+" Impossible d'enregistrer le token " + my_token)
return False, " Impossible de se connecter, merci de contacter le support technique"
except Exception as e:
exc_type, exc_obj, exc_tb = sys.exc_info()
mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno))
return False, "Impossible de se connecter, merci de contacter le support technique"