user_rows = self.db.get_list(
self.users_collection, {BaseTopic.id_field("users", user): user}
)
+ now = time()
user_content = None
- if user_rows:
- user_content = user_rows[0]
- salt = user_content["_admin"]["salt"]
- shadow_password = sha256(
- password.encode("utf-8") + salt.encode("utf-8")
- ).hexdigest()
- if shadow_password != user_content["password"]:
- user_content = None
+ if user:
+ user_rows = self.db.get_list(
+ self.users_collection,
+ {BaseTopic.id_field(self.users_collection, user): user},
+ )
+ if user_rows:
+ user_content = user_rows[0]
+ # Updating user_status for every system_admin id role login
+ mapped_roles = user_content.get("project_role_mappings")
+ for role in mapped_roles:
+ role_id = role.get("role")
+ role_assigned = self.db.get_one(
+ self.roles_collection,
+ {BaseTopic.id_field(self.roles_collection, role_id): role_id},
+ )
+
+ if role_assigned.get("permissions")["admin"]:
+ if role_assigned.get("permissions")["default"]:
+ if self.config.get("user_management"):
+ filt = {}
+ users = self.db.get_list(self.users_collection, filt)
+ for user_info in users:
+ if not user_info.get("username") == "admin":
+ if not user_info.get("_admin").get(
+ "account_expire_time"
+ ):
+ expire = now + 86400 * self.config.get(
+ "account_expire_days"
+ )
+ self.db.set_one(
+ self.users_collection,
+ {"_id": user_info["_id"]},
+ {"_admin.account_expire_time": expire},
+ )
+ else:
+ if now > user_info.get("_admin").get(
+ "account_expire_time"
+ ):
+ self.db.set_one(
+ self.users_collection,
+ {"_id": user_info["_id"]},
+ {"_admin.user_status": "expired"},
+ )
+ break
+
+ # To add "admin" user_status key while upgrading osm setup with feature enabled
+ if user_content.get("username") == "admin":
+ if self.config.get("user_management"):
+ self.db.set_one(
+ self.users_collection,
+ {"_id": user_content["_id"]},
+ {"_admin.user_status": "always-active"},
+ )
+
+ if not user_content.get("username") == "admin":
+ if self.config.get("user_management"):
+ if not user_content.get("_admin").get("account_expire_time"):
+ account_expire_time = now + 86400 * self.config.get(
+ "account_expire_days"
+ )
+ self.db.set_one(
+ self.users_collection,
+ {"_id": user_content["_id"]},
+ {"_admin.account_expire_time": account_expire_time},
+ )
+ else:
+ account_expire_time = user_content.get("_admin").get(
+ "account_expire_time"
+ )
+
+ if now > account_expire_time:
+ self.db.set_one(
+ self.users_collection,
+ {"_id": user_content["_id"]},
+ {"_admin.user_status": "expired"},
+ )
+ raise AuthException(
+ "Account expired", http_code=HTTPStatus.UNAUTHORIZED
+ )
+
+ if user_content.get("_admin").get("user_status") == "locked":
+ raise AuthException(
+ "Failed to login as the account is locked due to MANY FAILED ATTEMPTS"
+ )
+ elif user_content.get("_admin").get("user_status") == "expired":
+ raise AuthException(
+ "Failed to login as the account is expired"
+ )
+
+ salt = user_content["_admin"]["salt"]
+ shadow_password = sha256(
+ password.encode("utf-8") + salt.encode("utf-8")
+ ).hexdigest()
+ if shadow_password != user_content["password"]:
+ count = 1
+ if user_content.get("_admin").get("retry_count") >= 0:
+ count += user_content.get("_admin").get("retry_count")
+ self.db.set_one(
+ self.users_collection,
+ {"_id": user_content["_id"]},
+ {"_admin.retry_count": count},
+ )
+ self.logger.debug(
+ "Failed Authentications count: {}".format(count)
+ )
+
+ if user_content.get("username") == "admin":
+ user_content = None
+ else:
+ if not self.config.get("user_management"):
+ user_content = None
+ else:
+ if (
+ user_content.get("_admin").get("retry_count")
+ >= self.config["max_pwd_attempt"] - 1
+ ):
+ self.db.set_one(
+ self.users_collection,
+ {"_id": user_content["_id"]},
+ {"_admin.user_status": "locked"},
+ )
+ raise AuthException(
+ "Failed to login as the account is locked due to MANY FAILED ATTEMPTS"
+ )
+ else:
+ user_content = None
return user_content
def authenticate(self, credentials, token_info=None):
sleep(self.token_delay)
# user_content["_admin"]["last_token_time"] = now
# self.db.replace("users", user_content["_id"], user_content) # might cause race conditions
+ user_data = {
+ "_admin.last_token_time": now,
+ "_admin.retry_count": 0,
+ }
self.db.set_one(
self.users_collection,
{"_id": user_content["_id"]},
- {"_admin.last_token_time": now},
+ user_data,
)
token_id = "".join(
]
roles_list = [{"name": "project_admin", "id": rid}]
+ login_count = user_content.get("_admin").get("retry_count")
+ last_token_time = user_content.get("_admin").get("last_token_time")
+
+ admin_show = False
+ user_show = False
+ if self.config.get("user_management"):
+ for role in roles_list:
+ role_id = role.get("id")
+ permission = self.db.get_one(
+ self.roles_collection,
+ {BaseTopic.id_field(self.roles_collection, role_id): role_id},
+ )
+ if permission.get("permissions")["admin"]:
+ if permission.get("permissions")["default"]:
+ admin_show = True
+ break
+ else:
+ user_show = True
new_token = {
"issued_at": now,
"expires": now + 3600,
"user_id": user_content["_id"],
"admin": token_admin,
"roles": roles_list,
+ "login_count": login_count,
+ "last_login": last_token_time,
+ "admin_show": admin_show,
+ "user_show": user_show,
}
self.db.create(self.tokens_collection, new_token)
BaseTopic.format_on_new(user_info, make_public=False)
salt = uuid4().hex
user_info["_admin"]["salt"] = salt
+ user_info["_admin"]["user_status"] = "active"
present = time()
if not user_info["username"] == "admin":
- if self.config.get("pwd_expiry_check"):
- user_info["_admin"]["modified_time"] = present
- user_info["_admin"]["expire_time"] = present
+ if self.config.get("user_management"):
+ user_info["_admin"]["modified"] = present
+ user_info["_admin"]["password_expire_time"] = present
+ account_expire_time = present + 86400 * self.config.get(
+ "account_expire_days"
+ )
+ user_info["_admin"]["account_expire_time"] = account_expire_time
+
+ user_info["_admin"]["retry_count"] = 0
+ user_info["_admin"]["last_token_time"] = present
if "password" in user_info:
user_info["password"] = sha256(
user_info["password"].encode("utf-8") + salt.encode("utf-8")
).hexdigest()
+ user_info["_admin"]["password_history"] = {salt: user_info["password"]}
# "projects" are not stored any more
if "projects" in user_info:
del user_info["projects"]
"""
uid = user_info["_id"]
old_pwd = user_info.get("old_password")
+ unlock = user_info.get("unlock")
+ renew = user_info.get("renew")
+ permission_id = user_info.get("system_admin_id")
+
user_data = self.db.get_one(
self.users_collection, {BaseTopic.id_field("users", uid): uid}
)
raise AuthconnConflictException(
"Incorrect password", http_code=HTTPStatus.CONFLICT
)
+ # Unlocking the user
+ if unlock:
+ system_user = None
+ unlock_state = False
+ if not permission_id:
+ raise AuthconnConflictException(
+ "system_admin_id is the required field to unlock the user",
+ http_code=HTTPStatus.CONFLICT,
+ )
+ else:
+ system_user = self.db.get_one(
+ self.users_collection,
+ {
+ BaseTopic.id_field(
+ self.users_collection, permission_id
+ ): permission_id
+ },
+ )
+ mapped_roles = system_user.get("project_role_mappings")
+ for role in mapped_roles:
+ role_id = role.get("role")
+ role_assigned = self.db.get_one(
+ self.roles_collection,
+ {BaseTopic.id_field(self.roles_collection, role_id): role_id},
+ )
+ if role_assigned.get("permissions")["admin"]:
+ if role_assigned.get("permissions")["default"]:
+ user_data["_admin"]["retry_count"] = 0
+ user_data["_admin"]["user_status"] = "active"
+ unlock_state = True
+ break
+ if not unlock_state:
+ raise AuthconnConflictException(
+ "User '{}' does not have the privilege to unlock the user".format(
+ permission_id
+ ),
+ http_code=HTTPStatus.CONFLICT,
+ )
+ # Renewing the user
+ if renew:
+ system_user = None
+ renew_state = False
+ if not permission_id:
+ raise AuthconnConflictException(
+ "system_admin_id is the required field to renew the user",
+ http_code=HTTPStatus.CONFLICT,
+ )
+ else:
+ system_user = self.db.get_one(
+ self.users_collection,
+ {
+ BaseTopic.id_field(
+ self.users_collection, permission_id
+ ): permission_id
+ },
+ )
+ mapped_roles = system_user.get("project_role_mappings")
+ for role in mapped_roles:
+ role_id = role.get("role")
+ role_assigned = self.db.get_one(
+ self.roles_collection,
+ {BaseTopic.id_field(self.roles_collection, role_id): role_id},
+ )
+ if role_assigned.get("permissions")["admin"]:
+ if role_assigned.get("permissions")["default"]:
+ present = time()
+ account_expire = (
+ present + 86400 * self.config["account_expire_days"]
+ )
+ user_data["_admin"]["modified"] = present
+ user_data["_admin"]["account_expire_time"] = account_expire
+ user_data["_admin"]["user_status"] = "active"
+ renew_state = True
+ break
+ if not renew_state:
+ raise AuthconnConflictException(
+ "User '{}' does not have the privilege to renew the user".format(
+ permission_id
+ ),
+ http_code=HTTPStatus.CONFLICT,
+ )
BaseTopic.format_on_edit(user_data, user_info)
# User Name
usnm = user_info.get("username")
salt = uuid4().hex
if "_admin" not in user_data:
user_data["_admin"] = {}
+ if user_data.get("_admin").get("password_history"):
+ old_pwds = user_data.get("_admin").get("password_history")
+ else:
+ old_pwds = {}
+ for k, v in old_pwds.items():
+ shadow_password = sha256(
+ pswd.encode("utf-8") + k.encode("utf-8")
+ ).hexdigest()
+ if v == shadow_password:
+ raise AuthconnConflictException(
+ "Password is used before", http_code=HTTPStatus.CONFLICT
+ )
user_data["_admin"]["salt"] = salt
user_data["password"] = sha256(
pswd.encode("utf-8") + salt.encode("utf-8")
).hexdigest()
+ if len(old_pwds) >= 3:
+ old_pwds.pop(list(old_pwds.keys())[0])
+ old_pwds.update({salt: user_data["password"]})
+ user_data["_admin"]["password_history"] = old_pwds
if not user_data["username"] == "admin":
- if self.config.get("pwd_expiry_check"):
+ if self.config.get("user_management"):
present = time()
- if self.config.get("days"):
- expire = present + 86400 * self.config.get("days")
- user_data["_admin"]["modified_time"] = present
- user_data["_admin"]["expire_time"] = expire
+ if self.config.get("pwd_expire_days"):
+ expire = present + 86400 * self.config.get("pwd_expire_days")
+ user_data["_admin"]["modified"] = present
+ user_data["_admin"]["password_expire_time"] = expire
# Project-Role Mappings
# TODO: Check that user_info NEVER includes "project_role_mappings"
if "project_role_mappings" not in user_data:
projects / osm / NBI.git / blobdiff