Merge pull request #27 from Bhavneet-Sharma/release_3.1.0

Release version 3.1.0

ChangeLog.md

@@ -1,5 +1,8 @@
 # PyPowerStore Change Log
 
+## Version 3.1.0.0 - released on 29/02/24
+- Added support for session management using cookie.
+
 ## Version 3.0.0.0 - released on 31/01/24
 - Added support for creating, getting details, modifying and deleting
   a file interface.

PyPowerStore/__init__.py

@@ -3,6 +3,6 @@
 """__init__.py."""
 
 __title__ = 'PyPowerStore'
-__version__ = '3.0.0.0'
+__version__ = '3.1.0.0'
 __author__ = 'Dell Technologies or its subsidiaries'
-__copyright__ = 'Copyright 2019 Dell Technologies'
+__copyright__ = 'Copyright 2024 Dell Technologies'

PyPowerStore/client.py

@@ -1,12 +1,13 @@
 # -*- coding: utf-8 -*-
-# Copyright: (c) 2019, Dell Technologies
+# Copyright: (c) 2024, Dell Technologies
 
 """Client module for PowerStore"""
 
 import json
 import base64
 import socket
 import requests
+import time
 from requests.exceptions import SSLError
 from requests.exceptions import ConnectionError
 from requests.exceptions import TooManyRedirects
@@ -24,6 +25,106 @@
 ENGVIS_LIST = ["remote_support", "node", "volume_group", "remote_system"]
 
 
+class AuthenticationManager:
+    """Manage the powerstore authentication"""
+
+    def __init__(self, username, password, verify, application_type,
+                 timeout, host=None):
+        """ 
+        Initializes AuthenticationManager
+
+        :param username: array username
+        :type username: str
+        :param password: array password
+        :type password: str
+        :param verify: Whether the SSL cert will be verified
+        :type verify: bool
+        :param application_type: Application Type
+        :type application_type: str
+        :param timeout: How long to wait for the server to send data
+                        before giving up
+        :type timeout: float
+        :param host: (optional) IP of the Endpoint
+        :type host: str
+        """
+        self.username = username
+        self.password = password
+        self.verify = verify
+        self.application_type = application_type
+        self.timeout = timeout
+        self.host = host
+        self.dell_emc_token = None
+        self.cookie = None
+        self.idle_timeout = 0
+        self.creation_time = None
+        self.headers = {
+            'Accept': constants.APP_JSON,
+            'Accept-Language': constants.EN_US,
+            'content-type': constants.APP_JSON,
+            'Application-Type': self.application_type
+        }
+
+    def set_host(self, host):
+        """Set the host"""
+        self.host = host
+
+    def get_authorization(self):
+        """Get the authorization header"""
+        credentials = base64.b64encode(
+            "{username}:{password}".format(
+                username=self.username, password=self.password).encode())
+        return {'authorization': "Basic " + credentials.decode()}
+
+    def set_session_timeout_and_creation_time(self, login_response):
+        """Set the session timeout from login response object"""
+        if login_response.status_code == 200:
+            self.creation_time = time.time()
+            json_response = login_response.json()
+            login_data = login_response.json()[0] if isinstance(json_response, list) else {}
+            self.idle_timeout = login_data['idle_timeout'] if 'idle_timeout' in login_data else self.idle_timeout
+
+    def is_session_alive(self):
+        """Check if the session is alive or not"""
+        if self.creation_time and self.idle_timeout and \
+            ((time.time() - self.creation_time) < self.idle_timeout):
+                return True
+        return False
+
+    def login(self):
+        """Login to powerstore and set the token and cookie"""
+        login_url = constants.LOGIN_SESSION.format(self.host)
+        login_headers = dict(self.headers)
+        login_headers.update(self.get_authorization())
+        response = requests.request(
+            constants.GET, login_url, headers=login_headers, verify=self.verify,
+            timeout=self.timeout, params=constants.LOGIN_SESSION_DETAILS_QUERY)
+        self.set_session_timeout_and_creation_time(response)
+        self.dell_emc_token = response.headers.get('DELL-EMC-TOKEN')
+        self.cookie = response.cookies.get('auth_cookie')
+
+    def get_token_and_cookie(self):
+        """Get the DELL-EMC-TOKEN and set-cookie"""
+        auth_tokens = {}
+        if not self.dell_emc_token or not self.cookie or not self.is_session_alive():
+            self.login()
+
+        auth_tokens.update({'DELL-EMC-TOKEN': self.dell_emc_token})
+        auth_tokens.update({'Cookie': f'auth_cookie={self.cookie}'})
+        return auth_tokens
+
+    def logout_session(self):
+        """ Logout the current session """
+        login_url = constants.LOGOUT_URL.format(self.host)
+        logout_headers = {}
+        logout_headers.update(self.headers)
+        logout_headers.update({'DELL-EMC-TOKEN': self.dell_emc_token})
+        logout_headers.update({'Cookie': f'auth_cookie={self.cookie}'})
+        requests.request(
+            constants.POST, login_url, headers=logout_headers, verify=self.verify,
+            data=None, timeout=self.timeout)
+        self.dell_emc_token = None
+        self.cookie = None
+
 class Client():
     """Client class for PowerStore"""
     def __init__(self, username, password, verify, application_type,
@@ -51,6 +152,11 @@ def __init__(self, username, password, verify, application_type,
         self.application_type = application_type
         """Setting default timeout"""
         self.timeout = timeout if timeout else constants.TIMEOUT
+        self.auth_manager = AuthenticationManager(self.username,
+                                                  self.password,
+                                                  self.verify,
+                                                  self.application_type,
+                                                  self.timeout)
         LOG = helpers.get_logger(__name__, enable_log=enable_log)
 
     def fetch_response(self, http_method, url, payload=None, querystring=None,
@@ -78,8 +184,8 @@ def fetch_response(self, http_method, url, payload=None, querystring=None,
             'Application-Type': self.application_type
         }
         split_host = url.split('/')
-        auth_headers = {}
-        auth_headers = self.get_auth_token(split_host[2], headers)
+        self.auth_manager.set_host(split_host[2])
+        auth_headers = self.auth_manager.get_token_and_cookie()
 
         if split_host[5] in ENGVIS_LIST:
             headers['DELL-VISIBILITY'] = 'internal'
@@ -105,48 +211,8 @@ def fetch_response(self, http_method, url, payload=None, querystring=None,
             response = requests.request(
                 http_method, url, headers=headers, verify=self.verify,
                 timeout=self.timeout)
-        self.logout_session(split_host[2], headers)
         return response
 
-    def logout_session(self, host, headers):
-        """ Logout the current session.
-
-        :param host: IP of the host
-        :type: str
-        :param headers: The header for the https request
-        :type: dict
-        """
-
-        login_url = constants.LOGOUT_URL.format(host)
-        requests.request(
-            constants.POST, login_url, headers=headers, verify=self.verify,
-            data=None, timeout=self.timeout)
-
-    def get_auth_token(self, host, headers):
-        """ Logout the current session.
-
-        :param host: IP of the host
-        :type: str
-        :param headers: The header for the https request
-        :type: dict
-        :return: Dict containing authentication attributes
-        :rtype: dict
-        """
-        auth_tokens = {}
-        credentials = base64.b64encode(
-            "{username}:{password}".format(
-                username=self.username, password=self.password).encode())
-        headers.update({'authorization': "Basic " + credentials.decode()})
-        login_url = constants.LOGIN_SESSION.format(host)
-
-        response = requests.request(
-            constants.GET, login_url, headers=headers, verify=self.verify,
-            timeout=self.timeout)
-        if response:
-            auth_tokens.update({'DELL-EMC-TOKEN': response.headers.get('DELL-EMC-TOKEN')})
-            auth_tokens.update({'set-cookie': response.headers.get('set-cookie')})
-        return auth_tokens
-
     def is_valid_response(self, response):
         """ Check whether response is valid or not
 

PyPowerStore/tests/unit_tests/test_auth_manager.py

@@ -0,0 +1,55 @@
+import pytest
+import time
+import requests
+from unittest.mock import patch, MagicMock
+from PyPowerStore.client import AuthenticationManager
+
+class TestAuthenticationManager:
+    @pytest.fixture
+    def auth_manager(self):
+        return AuthenticationManager(username='test_user',
+                                     password='test_password',
+                                     verify=False,
+                                     application_type='test_app',
+                                     timeout=10)
+
+    def test_set_host(self, auth_manager):
+        auth_manager.set_host('test_host')
+        assert auth_manager.host == 'test_host'
+
+    def test_get_authorization(self, auth_manager):
+        auth_header = auth_manager.get_authorization()
+        assert auth_header == {'authorization': 'Basic dGVzdF91c2VyOnRlc3RfcGFzc3dvcmQ='}
+
+    def test_set_session_timeout_and_creation_time(self, auth_manager):
+        login_response_mock = MagicMock()
+        login_response_mock.status_code = 200
+        login_response_mock.json = MagicMock(return_value=[{'idle_timeout': 3600}])
+        auth_manager.set_session_timeout_and_creation_time(login_response_mock)
+        assert auth_manager.idle_timeout == 3600
+
+    def test_is_session_alive(self, auth_manager):
+        auth_manager.creation_time = 1
+        auth_manager.idle_timeout = 10
+        with patch.object(time,'time', return_value=5):
+            assert auth_manager.is_session_alive() == True
+
+    def test_login(self, auth_manager):
+        with patch.object(requests,'request'):
+            auth_manager.login()
+            assert auth_manager.dell_emc_token is not None
+            assert auth_manager.cookie is not None
+
+    def test_get_token_and_cookie(self, auth_manager):
+        with patch.object(requests,'request'):
+            auth_manager.login()
+            tokens = auth_manager.get_token_and_cookie()
+            assert tokens['DELL-EMC-TOKEN'] is not None
+            assert tokens['Cookie'] is not None
+
+    def test_logout_session(self, auth_manager):
+        with patch.object(requests,'request'):
+            auth_manager.login()
+            auth_manager.logout_session()
+            assert auth_manager.dell_emc_token is None
+            assert auth_manager.cookie is None

PyPowerStore/utils/constants.py

@@ -9,6 +9,8 @@
 PUT = 'PUT'
 DELETE = 'DELETE'
 PATCH = 'PATCH'
+APP_JSON = 'application/json'
+EN_US = "en-US"
 
 # Default Connection Timeout in seconds
 TIMEOUT = 120.0
@@ -484,6 +486,11 @@
               'virtual_machines'
 }
 
+# LOGIN_SESSION details
+LOGIN_SESSION_DETAILS_QUERY = {
+    'select': 'id,user_id,user,role_ids,is_password_change_required,is_built_in_user,user_type,idle_timeout'
+}
+
 # LDAP Account details
 LDAP_ACCOUNT_DETAILS_QUERY = {
     'select': 'id,role_id,domain_id,name,type,type_l10n,dn'

docs/conf.py

@@ -21,11 +21,11 @@
 # -- Project information -----------------------------------------------------
 
 project = 'PyPowerStore'
-copyright = '2022, Dell'
+copyright = '2024, Dell'
 author = 'Dell'
 
 # The full version, including alpha/beta/rc tags
-release = '3.0.0'
+release = '3.1.0.0'
 
 
 # -- General configuration ---------------------------------------------------

setup.py

@@ -10,7 +10,7 @@
 
 
 setup(name='PyPowerStore',
-      version='3.0.0.0',
+      version='3.1.0.0',
       description='Python Library for Dell PowerStore',
       author='Ansible Team at Dell',
       author_email='ansible.team@dell.com',