Merge pull request #161 from novonordisk-research/dont_verify

Optional verify

index.rst

@@ -96,6 +96,11 @@ sample set has been defined.
 Detailed Usage
 **************
 
+In general, properties that the user could be intercting with are exposed in
+`EmpowerHandler`. An example would be `project`, which the user could set to
+change which Empower project to log in to. Properties only exposed in 
+`EmpowerHandler.connection`, like `verify`, require more consideration to set.
+
 Class EmpowerHandler
 ==========================
 

src/OptiHPLCHandler/empower_api_core.py

@@ -1,7 +1,7 @@
 import getpass
 import logging
 import warnings
-from typing import Optional, Tuple
+from typing import Optional, Tuple, Union
 
 import keyring
 import requests
@@ -32,13 +32,15 @@ class EmpowerConnection:
     :ivar session_id: The session ID. None if not logged in.
     :ivar default_get_timeout: The default timeout to use for get requests.
     :ivar default_post_timeout: The default timeout to use for post requests.
+    :ivar verify: Whether to verify SSL certificates when connecting via HTTPS.
     """
 
     def __init__(
         self,
         address: str,
         project: Optional[str] = None,
         service: Optional[str] = None,
+        verify: Union[bool, str] = True,
     ) -> None:
         """
         Initialize the EmpowerConnection.
@@ -48,14 +50,21 @@ def __init__(
             is used.
         :param service: The service to use for logging in. If None, the first service in
             the list is used.
+        :param verify: Bool or string. If False, no verification of SSL certificates
+            is done when connecting via HTTPS. If it is a string, it should be the
+            path to the CA_BUNDLE file or directory with certificates of trusted CAs-
+            If true, the built-in list of trusted CAs will be used.
         """
         self.address = address.rstrip("/")  # Remove trailing slash if present
         self.username = getpass.getuser()
+        self.verify = verify
         if service is None:
             logger.debug("No service specified, getting service from Empower")
             try:
                 response = requests.get(
-                    self.address + "/authentication/db-service-list", timeout=10
+                    self.address + "/authentication/db-service-list",
+                    timeout=10,
+                    verify=self.verify,
                 )
             except requests.exceptions.Timeout as e:
                 timeout_string = f"Getting service from {self.address} timed out"
@@ -105,12 +114,12 @@ def login(
                 self.address + "/authentication/login",
                 json=body,
                 timeout=60,
+                verify=self.verify,
             )
         except requests.exceptions.Timeout as e:
             timeout_string = (
                 f"Login to {self.address} with username = {self.username} timed out"
             )
-            print(timeout_string)
             logger.error(timeout_string)
             raise requests.exceptions.Timeout(timeout_string) from e
         self.raise_for_status(response)
@@ -128,6 +137,7 @@ def logout(self) -> None:
             self.address + "/authentication/logout?sessionInfoID=" + self.session_id,
             headers=self.authorization_header,
             timeout=self.default_post_timeout,
+            verify=self.verify,
         )
         if response.status_code == 404:
             logger.debug(
@@ -152,14 +162,15 @@ def _requests_wrapper(
         :return: The results and message from the response.
         """
 
-        def _request_with_timeout(method, endpoint, header, body, timeout):
+        def _request_with_timeout(method, endpoint, header, body, timeout, verify):
             try:
                 return requests.request(
                     method,
                     endpoint,
                     json=body,
                     headers=header,
                     timeout=timeout,
+                    verify=verify,
                 )
             except requests.exceptions.Timeout as e:
                 timeout_string = f"{method}ing {body} to {endpoint} timed out"
@@ -172,13 +183,13 @@ def _request_with_timeout(method, endpoint, header, body, timeout):
         # Add slash between address and endpoint
         logger.debug("%sing %s to %s", method, body, address)
         response = _request_with_timeout(
-            method, address, self.authorization_header, body, timeout
+            method, address, self.authorization_header, body, timeout, self.verify
         )
         if response.status_code == 401:
             logger.debug("Token expired, logging in again")
             self.login()
             response = _request_with_timeout(
-                method, address, self.authorization_header, body, timeout
+                method, address, self.authorization_header, body, timeout, self.verify
             )
         logger.debug("Got response %s from %s", response.text, address)
         self.raise_for_status(response)

tests/test_core_api.py

@@ -15,6 +15,7 @@ def setUp(self, mock_requests) -> None:
             "results": [{"token": "test_token", "id": "test_id"}]
         }
         mock_response.status_code = 200
+        self.mock_response = mock_response
         mock_requests.post.return_value = mock_response
         # Since we log in, we need to mock that connection.
 
@@ -29,6 +30,13 @@ def setUp(self, mock_requests) -> None:
             service="test_service",
         )
         self.connection.login(username="test_username", password="test_password")
+        self.verify_connection = EmpowerConnection(
+            project="test_project",
+            address="https://test_address/",
+            service="test_service",
+            verify="test/CA/path",
+        )
+        self.verify_connection.login("", "")
 
     @patch("OptiHPLCHandler.empower_api_core.requests")
     def test_auto_service(self, mock_requests):
@@ -59,6 +67,38 @@ def test_login_timeout(self):
             self.connection.login(username="test_username", password="test_password")
             assert "timeout" in mock_post.call_args[1]
 
+    def test_verify_post(self):
+        with patch("OptiHPLCHandler.empower_api_core.requests.request") as mock_request:
+            self.connection.post("", {})
+            assert mock_request.call_args[1]["verify"] is True
+        with patch("OptiHPLCHandler.empower_api_core.requests.request") as mock_request:
+            self.verify_connection.post("", {})
+            assert mock_request.call_args[1]["verify"] == "test/CA/path"
+
+    def test_verify_get(self):
+        with patch("OptiHPLCHandler.empower_api_core.requests.request") as mock_request:
+            self.connection.get("")
+            assert mock_request.call_args[1]["verify"] is True
+        with patch("OptiHPLCHandler.empower_api_core.requests.request") as mock_request:
+            self.verify_connection.post("", {})
+            assert mock_request.call_args[1]["verify"] == "test/CA/path"
+
+    def test_verify_login(self):
+        with patch("OptiHPLCHandler.empower_api_core.requests.post") as mock_request:
+            self.connection.login("", "")
+            assert mock_request.call_args[1]["verify"] is True
+        with patch("OptiHPLCHandler.empower_api_core.requests.post") as mock_request:
+            self.verify_connection.login("", "")
+            assert mock_request.call_args[1]["verify"] == "test/CA/path"
+
+    def test_verify_logout(self):
+        with patch("OptiHPLCHandler.empower_api_core.requests.delete") as mock_request:
+            self.connection.logout()
+            assert mock_request.call_args[1]["verify"] is True
+        with patch("OptiHPLCHandler.empower_api_core.requests.delete") as mock_request:
+            self.verify_connection.logout()
+            assert mock_request.call_args[1]["verify"] == "test/CA/path"
+
     @patch("OptiHPLCHandler.empower_api_core.requests")
     def test_automatic_service_name(self, mock_requests):
         mock_response_service = MagicMock()