import uuid import secrets import hashlib import requests from sqlalchemy import or_ from datetime import timedelta from fastapi.exceptions import HTTPException from fastapi import status from databases import ( Users, People, Companies, UsersTokens, MongoQueryIdentity, ) from databases.no_sql_models.validations import ( PasswordHistoryViaUser, AccessHistoryViaUser, ) from api_library.date_time_actions.date_functions import system_arrow, client_arrow from api_configs import ApiStatic, Auth from api_services.redis.auth_actions.auth import save_access_token_to_redis class PasswordModule: @classmethod def generate_token(cls, length): return secrets.token_urlsafe(length) @classmethod def create_hashed_password(cls, domain, id_, password): salted_password = f"{domain}-{id_}-{password}" return hashlib.sha256(salted_password.encode()).hexdigest() @classmethod def check_hashed_password(cls, domain, id_, password, password_hashed): return cls.create_hashed_password(domain, id_, password) == password_hashed class AuthModule(PasswordModule): @classmethod def check_user_exits(cls, access_key, domain): found_user: Users = cls.filter_one( or_( cls.email == str(access_key).lower(), cls.phone_number == str(access_key).replace(" ", ""), ), ).data if not found_user: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Given access key or domain is not matching with the any user record.", ) other_domains_list = found_user.get_main_domain_and_other_domains( get_main_domain=False ) if domain not in other_domains_list: raise HTTPException( status_code=401, detail=dict(message="Unauthorized User attempts to connect api"), ) return found_user def generate_access_token(self): return self.generate_token(Auth.ACCESS_TOKEN_LENGTH) def remove_refresher_token(self, domain, disconnect: bool = False): if disconnect: registered_tokens = UsersTokens.filter_all( UsersTokens.user_id == self.id, system=True ) else: registered_tokens = UsersTokens.filter_all( UsersTokens.domain == domain, UsersTokens.user_id == self.id, system=True, ) registered_tokens.query.delete() UsersTokens.save() def check_password(self, password): main_domain = self.get_main_domain_and_other_domains(get_main_domain=True) if check_password := self.check_hashed_password( domain=main_domain, id_=self.uu_id, password_hashed=self.hash_password, password=password, ): return check_password raise HTTPException( status_code=401, detail="Password is not correct.", ) def check_password_is_different(self, password): main_domain = self.get_main_domain_and_other_domains(get_main_domain=True) if self.hash_password == self.create_hashed_password( domain=main_domain, id_=self.uu_id, password=password ): raise HTTPException( status_code=401, detail="New password is same with old password.", ) def create_password(self, password, password_token=None): if self.password_token: replace_day = 0 try: replace_day = int( str(self.password_expires_day or 0) .split(",")[0] .replace(" days", "") ) except Exception as e: err = e token_is_expired = system_arrow.now() >= system_arrow.get( self.password_expiry_begins ).shift(days=replace_day) if not password_token == self.password_token and token_is_expired: raise HTTPException( status_code=401, detail="Password token is not valid. Please request a new password token.", ) query_engine = MongoQueryIdentity(company_uuid=self.related_company) domain_via_user = query_engine.get_domain_via_user(user_uu_id=str(self.uu_id))[ "main_domain" ] new_password_dict = { "password": self.create_hashed_password( domain=domain_via_user, id_=self.uu_id, password=password ), "date": str(system_arrow.now()), } history_dict = PasswordHistoryViaUser( user_uu_id=str(self.uu_id), password_add=new_password_dict, access_history_detail={ "request": "", "ip": "", }, ) query_engine.refresh_password_history_via_user(payload=history_dict) self.password_expiry_begins = str(system_arrow.now()) self.hash_password = new_password_dict.get("password") if self.password_token: self.password_token = None self.save() def reset_password_token(self): self.password_expiry_begins = str(system_arrow.now()) self.password_token = self.generate_token(127) self.save() def generate_refresher_token(self, domain: str, remember_me=False): if remember_me: refresh_token = self.generate_token(Auth.REFRESHER_TOKEN_LENGTH) if already_token := UsersTokens.find_one( user_id=self.id, token_type="RememberMe", domain=domain ): already_token.update(token=refresh_token) already_token.expires_at = system_arrow.shift(days=3) already_token.save() return refresh_token UsersTokens.create( user_id=self.id, token_type="RememberMe", token=refresh_token, domain=domain, ) UsersTokens.save() return refresh_token return None def remainder_day(self): return float( timedelta( days=int( "".join( [ _ for _ in str(self.password_expires_day).split(",")[0] if _.isdigit() ] ) ) ).seconds ) class UserLoginModule(AuthModule): @classmethod def login_user_with_credentials(cls, data, request): found_user = Users.check_user_exits( access_key=data.access_key, domain=data.domain ) access_token = found_user.generate_access_token() query_engine = MongoQueryIdentity(company_uuid=found_user.related_company) if found_user.check_password(password=data.password): access_object_to_redis = save_access_token_to_redis( request=request, found_user=found_user, domain=data.domain, access_token=access_token, ) refresher_token = found_user.generate_refresher_token( domain=data.domain, remember_me=data.remember_me ) headers_request = request.headers headers_request = dict(headers_request) headers_request["evyos-user-agent"] = headers_request.get("user-agent") headers_request["evyos-platform"] = headers_request.get("user-agent") headers_request["evyos-ip-ext"] = "94.54.68.158" found_user.last_agent = headers_request.get("evyos-user-agent", None) found_user.last_platform = headers_request.get("evyos-platform", None) found_user.last_remote_addr = headers_request.get("evyos-ip-ext", None) found_user.last_seen = str(system_arrow.now()) if ext_ip := headers_request.get("evyos-ip-ext"): agent = headers_request.get("evyos-user-agent", "") platform = headers_request.get("evyos-platform", "") address = requests.get(f"http://ip-api.com/json/{ext_ip}").json() address_package = { "city": address["city"], "zip": address["zip"], "country": address["country"], "countryCode": address["countryCode"], "region": address["region"], "regionName": address["regionName"], } mongo_db = MongoQueryIdentity( company_uuid=str(found_user.related_company).replace(" ", ""), storage_reasoning="AccessHistory", ) filter_query = { "agent": agent, "platform": platform, "address": address_package, "user_id": found_user.id, } already_exits = mongo_db.mongo_engine.filter_by(filter_query) or None no_address_validates = mongo_db.mongo_engine.get_all()[0] == 0 record_id = uuid.uuid4().__str__() notice_link = ApiStatic.blacklist_login(record_id=record_id) found_people = People.find_one(id=found_user.person_id) access_via_user = query_engine.update_access_history_via_user( AccessHistoryViaUser( **{ "user_uu_id": found_user.uu_id.__str__(), "access_history": { "record_id": record_id, "agent": agent, "platform": platform, "address": address_package, "ip": ext_ip, "access_token": access_token, "created_at": system_arrow.now().timestamp(), # "is_confirmed": True if no_address_validates else False, # "is_first": True if no_address_validates else False, }, } ) ) if already_exits: update_mongo = mongo_db.mongo_engine.table.update_one( filter=filter_query, update={ "$set": { "ip": ext_ip, "access_token": access_token, "created_at": system_arrow.now().timestamp(), } }, ) else: mongo_db.mongo_engine.insert( payload={ "user_id": found_user.id, "record_id": record_id, "agent": agent, "platform": platform, "address": address_package, "ip": ext_ip, "access_token": access_token, "created_at": system_arrow.now().timestamp(), "is_confirmed": True if no_address_validates else False, "is_first": True if no_address_validates else False, } ) found_user.remember_me = bool(data.remember_me) found_user.save() return { "access_token": access_token, "refresher_token": refresher_token, "user": found_user, "access_object": access_object_to_redis, } raise HTTPException( status_code=401, detail="Login is not successful. Please check your credentials.", ) # UserLogger.log_error( # dict( # user_id=found_user.id, # domain=data.domain, # access_key=data.access_key, # agent=found_user.last_agent, # ip=getattr(request, "remote_addr", None) # or request.headers.get("X-Forwarded-For", None), # platform=found_user.last_platform, # login_date=str(DateTimeLocal.now()), # is_login=True, # ) # ) # if ( # not str(found_people.country_code).lower() # == str(address_package.get("countryCode")).lower() # ): # send_email_completed = send_email( # subject=f"Dear {found_user.nick_name}, your password has been changed.", # receivers=[str(found_user.email)], # html=invalid_ip_or_address_found( # user_name=found_user.nick_name, # address=address_package, # notice_link=notice_link, # ), # ) # if not send_email_completed: # raise HTTPException( # status_code=400, # detail="An error occured at sending email. Please contact with support team.", # )