Expose TLS verification

ansys/rep/client/auth/authenticate.py

@@ -26,21 +26,41 @@ def authenticate(
     password: str = None,
     refresh_token: str = None,
     timeout: float = 10.0,
+    verify: bool | str = True,
     **kwargs,
 ):
     """
     Authenticate user with either password or refresh token against REP authentication service.
     If successful, the response includes access and refresh tokens.
 
-    Args:
-        url (str): The base path for the server to call, e.g. "https://127.0.0.1:8443/rep".
-        username (str): Username
-        password (str): Password
-        refresh_token (str, optional): Refresh token.
-        timeout (float, optional): Timeout in seconds. Defaults to 10.
-        scope (str, optional): String containing one or more requested scopes.
-                               Defaults to 'openid'.
-        client_id (str, optional): The client type. Defaults to 'rep-cli'.
+    Parameters
+    ----------
+
+    url : str
+        The base path for the server to call, e.g. "https://127.0.0.1:8443/rep".
+    realm : str
+        Keycloak realm, defaults to 'rep'.
+    grant_type: str
+        Authentication method, defaults to 'password'.
+    username : str
+        Username
+    password : str
+        Password
+    refresh_token : str, optional
+        Refresh token.
+    timeout : float, optional
+        Timeout in seconds. Defaults to 10.
+    scope : str, optional
+        String containing one or more requested scopes. Defaults to 'openid'.
+    client_id : str, optional
+        The client type. Defaults to 'rep-cli'.
+    client_secret : str, optional
+        The client secret.
+    verify: bool | str, optional
+        Either a boolean, in which case it controls whether we verify the
+        server's TLS certificate, or a string, in which case it must be
+        a path to a CA bundle to use.
+        See the :class:`requests.Session` doc.
 
     Returns:
         dict: JSON-encoded content of a :class:`requests.Response`
@@ -54,7 +74,7 @@ def authenticate(
     log.debug(f"Authenticating using {auth_url}")
 
     session = requests.Session()
-    session.verify = False
+    session.verify = verify
     session.headers = ({"content-type": "application/x-www-form-urlencoded"},)
 
     token_url = f"{auth_url}/protocol/openid-connect/token"

ansys/rep/client/client.py

@@ -49,6 +49,14 @@ class Client(object):
         If True, the query parameter ``fields="all"`` is applied by default
         to all requests, so that all available fields are returned for
         the requested resources.
+    verify: bool | str, optional
+        Either a boolean, in which case it controls whether we verify the
+        server's TLS certificate, or a string, in which case it must be
+        a path to a CA bundle to use. Defaults to False.
+        See the :class:`requests.Session` documentation for more details.
+    disable_insecure_warnings: bool, optional
+        Disable warnings about insecure HTTPS requests. Defaults to True.
+        See urllib3 documentation about TLS Warnings for more details.
 
     Examples
     --------
@@ -85,6 +93,8 @@ def __init__(
         refresh_token: str = None,
         auth_url: str = None,
         all_fields=True,
+        verify: bool | str = False,
+        disable_insecure_warnings: bool = True,
     ):
 
         self.rep_url = rep_url
@@ -98,6 +108,7 @@ def __init__(
         self.scope = scope
         self.client_id = client_id
         self.client_secret = client_secret
+        self.verify = verify
 
         if access_token:
             log.debug("Authenticate with access token")
@@ -122,12 +133,17 @@ def __init__(
                 username=username,
                 password=password,
                 refresh_token=refresh_token,
+                verify=self.verify,
             )
             self.access_token = tokens["access_token"]
             # client credentials flow does not return a refresh token
             self.refresh_token = tokens.get("refresh_token", None)
 
-        self.session = create_session(self.access_token)
+        self.session = create_session(
+            self.access_token,
+            verify=verify,
+            disable_insecure_warnings=disable_insecure_warnings,
+        )
         if all_fields:
             self.session.params = {"fields": "all"}
 
@@ -167,6 +183,7 @@ def refresh_access_token(self):
                 scope=self.scope,
                 client_id=self.client_id,
                 client_secret=self.client_secret,
+                verify=self.verify,
             )
         else:
             # Other workflows for authentication generally support refresh_tokens
@@ -179,6 +196,7 @@ def refresh_access_token(self):
                 client_secret=self.client_secret,
                 username=self.username,
                 refresh_token=self.refresh_token,
+                verify=self.verify,
             )
         self.access_token = tokens["access_token"]
         self.refresh_token = tokens.get("refresh_token", None)

ansys/rep/client/connection.py

@@ -13,22 +13,39 @@
 log = logging.getLogger(__name__)
 
 
-def create_session(access_token: str = None) -> requests.Session:
+def create_session(
+    access_token: str = None,
+    verify: bool | str = True,
+    disable_insecure_warnings=False,
+) -> requests.Session:
     """Returns a :class:`requests.Session` object configured for REP with given access token
 
-    Args:
-        access_token (str): The access token provided by :meth:`ansys.rep.client.auth.authenticate`
+    Parameters
+    ----------
+    access_token : str
+        The access token provided by :meth:`ansys.rep.client.auth.authenticate`
 
-    Returns:
-        :class:`requests.Session`: The session object.
+    Returns
+    -------
+    :class:`requests.Session`
+        The session object.
+    verify: bool | str, optional
+        Either a boolean, in which case it controls whether we verify the
+        server's TLS certificate, or a string, in which case it must be
+        a path to a CA bundle to use.
+        See the :class:`requests.Session` doc.
+    disable_insecure_warnings: bool, optional
+        Disable warnings about insecure HTTPS requests.
     """
     session = requests.Session()
 
     # Disable SSL certificate verification and warnings about it
-    session.verify = False
-    requests.packages.urllib3.disable_warnings(
-        requests.packages.urllib3.exceptions.InsecureRequestWarning
-    )
+    session.verify = verify
+
+    if disable_insecure_warnings:
+        requests.packages.urllib3.disable_warnings(
+            requests.packages.urllib3.exceptions.InsecureRequestWarning
+        )
 
     # Set basic content type to json
     session.headers = {

tests/auth/test_api.py

@@ -95,6 +95,7 @@ def test_impersonate_user(self):
                 subject_token=client.access_token,
                 requested_token_type="urn:ietf:params:oauth:token-type:refresh_token",
                 requested_subject=new_user.id,
+                verify=False,
             )
         except REPError as e:
             if e.response.status_code == 501 and "Feature not enabled" in e.reason:

tests/auth/test_authenticate.py

@@ -7,6 +7,8 @@
 # ----------------------------------------------------------
 import logging
 
+import requests
+
 from ansys.rep.client.auth import authenticate
 from tests.rep_test import REPTestCase
 
@@ -15,7 +17,17 @@
 
 class AuthenticationTest(REPTestCase):
     def test_authenticate(self):
-        resp = authenticate(url=self.rep_url, username=self.username, password=self.password)
+        resp = authenticate(
+            url=self.rep_url, username=self.username, password=self.password, verify=False
+        )
 
         self.assertIn("access_token", resp)
         self.assertIn("refresh_token", resp)
+
+    def test_authenticate_with_tls_verification(self):
+
+        with self.assertRaises(requests.exceptions.SSLError) as context:
+            _ = authenticate(
+                url=self.rep_url, username=self.username, password=self.password, verify=True
+            )
+        self.assertTrue("CERTIFICATE_VERIFY_FAILED" in str(context.exception))

tests/test_connection.py

@@ -17,10 +17,12 @@
 class ConnectionTest(REPTestCase):
     def test_connection(self):
         rep_url = self.rep_url
-        resp = authenticate(url=rep_url, username=self.username, password=self.password)
+        resp = authenticate(
+            url=rep_url, username=self.username, password=self.password, verify=False
+        )
         access_token = resp["access_token"]
 
-        with create_session(access_token) as session:
+        with create_session(access_token, verify=False, disable_insecure_warnings=True) as session:
             jms_api_url = f"{rep_url}/jms/api/v1"
             log.info(f"Ping {jms_api_url}")
             ping(session, jms_api_url)