+
+ def generate_otp(self):
+ otp = "".join(random_choice("0123456789") for i in range(0, 4))
+ return otp
+
+ def send_email(self, indata):
+ user = indata.get("username")
+ user_rows = self.db.get_list(self.users_collection, {"username": user})
+ sender_password = None
+ otp_expiry_time = self.config.get("otp_expiry_time", 300)
+ if not re.match(email_schema["pattern"], indata.get("email_id")):
+ raise AuthException(
+ "Invalid email-id",
+ http_code=HTTPStatus.BAD_REQUEST,
+ )
+ if self.config.get("sender_email"):
+ sender_email = self.config["sender_email"]
+ else:
+ raise AuthException(
+ "sender_email not found",
+ http_code=HTTPStatus.NOT_FOUND,
+ )
+ if self.config.get("smtp_server"):
+ smtp_server = self.config["smtp_server"]
+ else:
+ raise AuthException(
+ "smtp server not found",
+ http_code=HTTPStatus.NOT_FOUND,
+ )
+ if self.config.get("smtp_port"):
+ smtp_port = self.config["smtp_port"]
+ else:
+ raise AuthException(
+ "smtp port not found",
+ http_code=HTTPStatus.NOT_FOUND,
+ )
+ sender_password = self.config.get("sender_password") or None
+ if user_rows:
+ user_data = user_rows[0]
+ user_status = user_data["_admin"]["user_status"]
+ if not user_data.get("project_role_mappings", None):
+ raise AuthException(
+ "can't find a default project for this user",
+ http_code=HTTPStatus.UNAUTHORIZED,
+ )
+ if user_status != "active" and user_status != "always-active":
+ raise AuthException(
+ f"User account is {user_status}.Please contact the system administrator.",
+ http_code=HTTPStatus.UNAUTHORIZED,
+ )
+ if user_data.get("email_id"):
+ if user_data["email_id"] == indata.get("email_id"):
+ otp = self.generate_otp()
+ encode_otp = (
+ sha256(
+ otp.encode("utf-8")
+ + user_data["_admin"]["salt"].encode("utf-8")
+ )
+ ).hexdigest()
+ otp_field = {encode_otp: time() + otp_expiry_time, "retries": 0}
+ user_data["OTP"] = otp_field
+ uid = user_data["_id"]
+ idf = BaseTopic.id_field("users", uid)
+ reciever_email = user_data["email_id"]
+ email_template_path = self.config.get("email_template")
+ with open(email_template_path, "r") as et:
+ email_template = et.read()
+ msg = EmailMessage()
+ msg = MIMEMultipart("alternative")
+ html_content = email_template.format(
+ username=user_data["username"],
+ otp=otp,
+ validity=otp_expiry_time // 60,
+ )
+ html = MIMEText(html_content, "html")
+ msg["Subject"] = "OSM password reset request"
+ msg.attach(html)
+ with smtplib.SMTP(smtp_server, smtp_port) as smtp:
+ smtp.starttls()
+ if sender_password:
+ smtp.login(sender_email, sender_password)
+ smtp.sendmail(sender_email, reciever_email, msg.as_string())
+ self.db.set_one(self.users_collection, {idf: uid}, user_data)
+ return {"email": "sent"}
+ else:
+ raise AuthException(
+ "No email id is registered for this user.Please contact the system administrator.",
+ http_code=HTTPStatus.NOT_FOUND,
+ )
+ else:
+ raise AuthException(
+ "user not found",
+ http_code=HTTPStatus.NOT_FOUND,
+ )
+
+ def validate_otp(self, indata):
+ otp = indata.get("otp")
+ user = indata.get("username")
+ user_rows = self.db.get_list(self.users_collection, {"username": user})
+ user_data = user_rows[0]
+ uid = user_data["_id"]
+ idf = BaseTopic.id_field("users", uid)
+ retry_count = self.config.get("retry_count", 3)
+ if user_data:
+ salt = user_data["_admin"]["salt"]
+ actual_otp = sha256(otp.encode("utf-8") + salt.encode("utf-8")).hexdigest()
+ if not user_data.get("OTP"):
+ otp_field = {"retries": 1}
+ user_data["OTP"] = otp_field
+ self.db.set_one(self.users_collection, {idf: uid}, user_data)
+ return {"retries": user_data["OTP"]["retries"]}
+ for key, value in user_data["OTP"].items():
+ curr_time = time()
+ if key == actual_otp and curr_time < value:
+ user_data["OTP"] = {}
+ self.db.set_one(self.users_collection, {idf: uid}, user_data)
+ return {"valid": "True", "password_change": "True"}
+ else:
+ user_data["OTP"]["retries"] += 1
+ self.db.set_one(self.users_collection, {idf: uid}, user_data)
+ if user_data["OTP"].get("retries") >= retry_count:
+ raise AuthException(
+ "Invalid OTP. Maximum retries exceeded",
+ http_code=HTTPStatus.TOO_MANY_REQUESTS,
+ )
+ return {"retry_count": user_data["OTP"]["retries"]}
projects / osm / NBI.git / commitdiff