X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_nbi%2Fauth.py;h=0b3264fb37ecd6bc9355373e506befa963ca5004;hb=9af2a4785d3a77772fd205aa572cc6a64d4d1003;hp=6c9ee925abaa3b6721604adbcd24a2f253de45de;hpb=b6480fc3be0d05e4e64f35a12dccdabf249ed94e;p=osm%2FNBI.git

diff --git a/osm_nbi/auth.py b/osm_nbi/auth.py
index 6c9ee92..0b3264f 100644
--- a/osm_nbi/auth.py
+++ b/osm_nbi/auth.py
@@ -283,7 +283,7 @@ class Authenticator:
             (r for r in records if r["name"] == "system_admin"), None
         ):
             with open(self.roles_to_operations_file, "r") as stream:
-                roles_to_operations_yaml = yaml.load(stream, Loader=yaml.Loader)
+                roles_to_operations_yaml = yaml.safe_load(stream)
 
             role_names = []
             for role_with_operations in roles_to_operations_yaml["roles"]:
@@ -449,9 +449,11 @@ class Authenticator:
                 elif auth_list[0].lower() == "basic":
                     user_passwd64 = auth_list[-1]
             if not token:
-                if cherrypy.session.get("Authorization"):
+                if cherrypy.session.get("Authorization"):  # pylint: disable=E1101
                     # 2. Try using session before request a new token. If not, basic authentication will generate
-                    token = cherrypy.session.get("Authorization")
+                    token = cherrypy.session.get(  # pylint: disable=E1101
+                        "Authorization"
+                    )
                     if token == "logout":
                         token = None  # force Unauthorized response to insert user password again
                 elif user_passwd64 and cherrypy.request.config.get(
@@ -466,10 +468,10 @@ class Authenticator:
                     except Exception:
                         pass
                     outdata = self.new_token(
-                        None, {"username": user, "password": passwd}
+                        None, {"username": user, "password": passwd}, None
                     )
                     token = outdata["_id"]
-                    cherrypy.session["Authorization"] = token
+                    cherrypy.session["Authorization"] = token  # pylint: disable=E1101
 
             if not token:
                 raise AuthException(
@@ -502,14 +504,14 @@ class Authenticator:
                     query_string_operations,
                     item_id,
                 )
-                self.logger.info("RBAC_auth: {}", format(RBAC_auth))
+                self.logger.info("RBAC_auth: {}".format(RBAC_auth))
                 token_info["allow_show_user_project_role"] = RBAC_auth
 
             return token_info
         except AuthException as e:
             if not isinstance(e, AuthExceptionUnauthorized):
-                if cherrypy.session.get("Authorization"):
-                    del cherrypy.session["Authorization"]
+                if cherrypy.session.get("Authorization"):  # pylint: disable=E1101
+                    del cherrypy.session["Authorization"]  # pylint: disable=E1101
                 cherrypy.response.headers[
                     "WWW-Authenticate"
                 ] = 'Bearer realm="{}"'.format(e)
@@ -768,3 +770,26 @@ class Authenticator:
         else:
             self.tokens_cache.clear()
         self.msg.write("admin", "revoke_token", {"_id": token} if token else None)
+
+    def check_password_expiry(self, outdata):
+        """
+        This method will check for password expiry of the user
+        :param outdata: user token information
+        """
+        user_content = None
+        present_time = time()
+        user = outdata["username"]
+        if self.config["authentication"].get("pwd_expiry_check"):
+            user_content = self.db.get_list("users", {"username": user})[0]
+            if not user_content.get("username") == "admin":
+                user_content["_admin"]["modified_time"] = present_time
+                if user_content.get("_admin").get("expire_time"):
+                    expire_time = user_content["_admin"]["expire_time"]
+                else:
+                    expire_time = present_time
+                uid = user_content["_id"]
+                self.db.set_one("users", {"_id": uid}, user_content)
+                if not present_time < expire_time:
+                    return True
+        else:
+            pass