Feature 11034: Forgot Password in OSM
Change-Id: I7df89b691f994a4bdf089f1a2677ab61f46b6838
Signed-off-by: jegan <jegan.s@tataelxsi.co.in>
Signed-off-by: garciadeblas <gerardo.garciadeblas@telefonica.com>
diff --git a/osm_nbi/authconn_internal.py b/osm_nbi/authconn_internal.py
index 94e6e47..cf8c55a 100644
--- a/osm_nbi/authconn_internal.py
+++ b/osm_nbi/authconn_internal.py
@@ -41,13 +41,17 @@
from osm_common.dbbase import DbException
from osm_nbi.base_topic import BaseTopic
from osm_nbi.utils import cef_event, cef_event_builder
-from osm_nbi.validation import is_valid_uuid
+from osm_nbi.validation import is_valid_uuid, email_schema
from time import time, sleep
from http import HTTPStatus
from uuid import uuid4
from hashlib import sha256
from copy import deepcopy
from random import choice as random_choice
+import smtplib
+from email.message import EmailMessage
+from email.mime.text import MIMEText
+from email.mime.multipart import MIMEMultipart
class AuthconnInternal(Authconn):
@@ -147,7 +151,7 @@
self.logger.exception(exmsg)
raise AuthException(exmsg, http_code=HTTPStatus.UNAUTHORIZED)
- def validate_user(self, user, password):
+ def validate_user(self, user, password, otp=None):
"""
Validate username and password via appropriate backend.
:param user: username of the user.
@@ -246,7 +250,8 @@
raise AuthException(
"Failed to login as the account is expired"
)
-
+ if otp:
+ return user_content
salt = user_content["_admin"]["salt"]
shadow_password = sha256(
password.encode("utf-8") + salt.encode("utf-8")
@@ -309,9 +314,12 @@
user = credentials.get("username")
password = credentials.get("password")
project = credentials.get("project_id")
+ otp_validation = credentials.get("otp")
# Try using username/password
- if user:
+ if otp_validation:
+ user_content = self.validate_user(user, password=None, otp=otp_validation)
+ elif user:
user_content = self.validate_user(user, password)
if not user_content:
cef_event(
@@ -657,8 +665,11 @@
BaseTopic.format_on_edit(user_data, user_info)
# User Name
usnm = user_info.get("username")
+ email_id = user_info.get("email_id")
if usnm:
user_data["username"] = usnm
+ if email_id:
+ user_data["email_id"] = email_id
# If password is given and is not already encripted
pswd = user_info.get("password")
if pswd and (
@@ -858,3 +869,129 @@
{BaseTopic.id_field("projects", project_id): project_id},
project_info,
)
+
+ 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"]}