feat: add function to get the proxy details

solnlib/conf_manager.py

@@ -21,28 +21,27 @@
 import json
 import logging
 import traceback
-from typing import List
+from typing import List, Union, Dict, NoReturn
 
 from splunklib import binding, client
 
 from . import splunk_rest_client as rest_client
 from .credentials import CredentialManager, CredentialNotExistException
 from .utils import retry
+from .net_utils import is_valid_port, is_valid_hostname
+from .soln_exceptions import (
+    ConfManagerException,
+    ConfStanzaNotExistException,
+    InvalidPortError,
+    InvalidHostnameError,
+)
 
 __all__ = [
-    "ConfStanzaNotExistException",
     "ConfFile",
-    "ConfManagerException",
     "ConfManager",
 ]
 
 
-class ConfStanzaNotExistException(Exception):
-    """Exception raised by ConfFile class."""
-
-    pass
-
-
 class ConfFile:
     """Configuration file."""
 
@@ -361,12 +360,6 @@ def reload(self):
         self._conf.get("_reload")
 
 
-class ConfManagerException(Exception):
-    """Exception raised by ConfManager class."""
-
-    pass
-
-
 class ConfManager:
     """Configuration file manager.
 
@@ -557,3 +550,68 @@ def get_log_level(
             f"taking {default_log_level} as log level."
         )
         return default_log_level
+
+
+def get_proxy_dict(
+    logger: logging.Logger,
+    session_key: str,
+    app_name: str,
+    conf_name: str,
+    proxy_stanza: str = "proxy",
+    **kwargs,
+) -> Union[Dict[str, str], NoReturn]:
+    """This function returns the log level for the addon from configuration
+    file.
+
+    Arguments:
+        logger: Logger.
+        session_key: Splunk access token.
+        app_name: Add-on name.
+        conf_name: Configuration file name where logging stanza is.
+        proxy_stanza: Proxy stanza that would contain the Proxy details
+    Returns:
+        A dictionary is returned with stanza details present in the file.
+        The keys related to `eai` are removed before returning.
+
+    Examples:
+        >>> from solnlib import conf_manager
+        >>> proxy_details = conf_manager.get_proxy_dict(
+        >>>     logger,
+        >>>     "session_key",
+        >>>     "ADDON_NAME",
+        >>>     "splunk_ta_addon_settings",
+        >>> )
+    """
+    proxy_dict = {}
+    try:
+        cfm = ConfManager(
+            session_key,
+            app_name,
+            realm=f"__REST_CREDENTIAL__#{app_name}#configs/conf-{conf_name}",
+        )
+        conf = cfm.get_conf(conf_name)
+    except ConfManagerException:
+        logger.error(f"Failed to fetch configuration file '{conf_name}'.")
+    else:
+        try:
+            proxy_dict = conf.get(proxy_stanza)
+        except ConfStanzaNotExistException:
+            logger.error(
+                f"Failed to fetch '{proxy_stanza}' from the configuration file '{conf_name}'. "
+            )
+        else:
+            # remove the other fields that are added by ConfFile class
+            proxy_dict.pop("disabled", None)
+            proxy_dict.pop("eai:access", None)
+            proxy_dict.pop("eai:appName", None)
+            proxy_dict.pop("eai:userName", None)
+
+            if "proxy_port" in kwargs:
+                if not is_valid_port(proxy_dict.get(kwargs["proxy_port"])):
+                    logger.error("Invalid proxy port provided.")
+                    raise InvalidPortError("The provided port is not valid.")
+            if "proxy_host" in kwargs:
+                if not is_valid_hostname(proxy_dict.get(kwargs["proxy_host"])):
+                    logger.error("Invalid proxy host provided.")
+                    raise InvalidHostnameError("The provided hostname is not valid.")
+    return proxy_dict

solnlib/soln_exceptions.py

@@ -0,0 +1,37 @@
+#
+# Copyright 2024 Splunk Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+class ConfManagerException(Exception):
+    """Exception raised by ConfManager class."""
+
+    pass
+
+
+class ConfStanzaNotExistException(Exception):
+    """Exception raised by ConfFile class."""
+
+    pass
+
+
+class InvalidPortError(ValueError):
+    """Exception raised when an invalid proxy port is provided."""
+
+    pass
+
+
+class InvalidHostnameError(ValueError):
+    """Exception raised when an invalid proxy hostname is provided."""
+
+    pass

tests/integration/data/solnlib_demo/default/splunk_ta_addon_settings.conf

@@ -1,2 +1,11 @@
 [logging]
 log_level = DEBUG
+
+[proxy]
+proxy_enabled =
+proxy_type = http
+proxy_url = remote_host
+proxy_port = 3128
+proxy_username =
+proxy_password =
+proxy_rdns =

tests/integration/data/solnlib_demo/default/splunk_ta_addon_settings_invalid.conf

@@ -0,0 +1,9 @@
+
+[invalid_proxy]
+proxy_enabled =
+proxy_type = http3
+proxy_url = remote:host:invalid
+proxy_port = 99999
+proxy_username =
+proxy_password =
+proxy_rdns =

tests/integration/test_conf_manager.py

@@ -15,14 +15,20 @@
 #
 
 import context
-import os.path as op
-import sys
 import pytest
-from solnlib import conf_manager
+from solnlib import conf_manager, soln_exceptions
 from unittest import mock
 
 
-sys.path.insert(0, op.dirname(op.dirname(op.abspath(__file__))))
+VALID_PROXY_DICT = {
+    "proxy_enabled": None,
+    "proxy_type": "http",
+    "proxy_url": "remote_host",
+    "proxy_port": "3128",
+    "proxy_username": None,
+    "proxy_password": None,
+    "proxy_rdns": None,
+}
 
 
 def _build_conf_manager(session_key: str) -> conf_manager.ConfManager:
@@ -40,7 +46,7 @@ def test_conf_manager_when_no_conf_then_throw_exception():
     session_key = context.get_session_key()
     cfm = _build_conf_manager(session_key)
 
-    with pytest.raises(conf_manager.ConfManagerException):
+    with pytest.raises(soln_exceptions.ConfManagerException):
         cfm.get_conf("non_existent_configuration_file")
 
 
@@ -50,7 +56,7 @@ def test_conf_manager_when_conf_file_exists_but_no_specific_stanza_then_throw_ex
 
     splunk_ta_addon_settings_conf_file = cfm.get_conf("splunk_ta_addon_settings")
 
-    with pytest.raises(conf_manager.ConfStanzaNotExistException):
+    with pytest.raises(soln_exceptions.ConfStanzaNotExistException):
         splunk_ta_addon_settings_conf_file.get(
             "non_existent_stanza_under_existing_conf_file"
         )
@@ -60,6 +66,7 @@ def test_conf_manager_when_conf_file_exists_but_no_specific_stanza_then_throw_ex
     "stanza_name,expected_result",
     [
         ("logging", True),
+        ("proxy", True),
         ("non_existent_stanza_under_existing_conf_file", False),
     ],
 )
@@ -109,7 +116,7 @@ def test_conf_manager_delete_non_existent_stanza_then_throw_exception():
 
     splunk_ta_addon_settings_conf_file = cfm.get_conf("splunk_ta_addon_settings")
 
-    with pytest.raises(conf_manager.ConfStanzaNotExistException):
+    with pytest.raises(soln_exceptions.ConfStanzaNotExistException):
         splunk_ta_addon_settings_conf_file.delete(
             "non_existent_stanza_under_existing_conf_file"
         )
@@ -164,3 +171,43 @@ def test_get_log_level_incorrect_log_level_field():
     )
 
     assert expected_log_level == log_level
+
+
+def test_get_proxy_dict():
+    session_key = context.get_session_key()
+    expected_proxy_dict = VALID_PROXY_DICT
+    proxy_dict = conf_manager.get_proxy_dict(
+        logger=mock.MagicMock(),
+        session_key=session_key,
+        app_name="solnlib_demo",
+        conf_name="splunk_ta_addon_settings",
+    )
+    assert expected_proxy_dict == proxy_dict
+
+
+def test_invalid_proxy_port():
+    session_key = context.get_session_key()
+
+    with pytest.raises(soln_exceptions.InvalidPortError):
+        conf_manager.get_proxy_dict(
+            logger=mock.MagicMock(),
+            session_key=session_key,
+            app_name="solnlib_demo",
+            conf_name="splunk_ta_addon_settings_invalid",
+            proxy_stanza="invalid_proxy",
+            proxy_port="proxy_port",
+        )
+
+
+def test_invalid_proxy_host():
+    session_key = context.get_session_key()
+
+    with pytest.raises(soln_exceptions.InvalidHostnameError):
+        conf_manager.get_proxy_dict(
+            logger=mock.MagicMock(),
+            session_key=session_key,
+            app_name="solnlib_demo",
+            conf_name="splunk_ta_addon_settings_invalid",
+            proxy_stanza="invalid_proxy",
+            proxy_host="proxy_url",
+        )