''' 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 import GlobalVariable as MYSY_GV 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 = MYSY_GV.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"