feat(mtls): implement of mutual TLS communication for fetching server info.

.github/workflows/build-test-release.yml

@@ -119,23 +119,23 @@ jobs:
           export SPLUNK_HOME=/opt/splunk
           wget -qO /tmp/splunk.tgz "${SPLUNK_BUILD_URL}"
           sudo tar -C /opt -zxf /tmp/splunk.tgz
-          sudo cp -r tests/integration/data/solnlib_demo $SPLUNK_HOME/etc/apps
-          sudo cp -r solnlib $SPLUNK_HOME/etc/apps/solnlib_demo/bin/
-          sudo mkdir -p $SPLUNK_HOME/etc/apps/Splunk_TA_test/default/
-          sudo chown -R "$USER":"$USER" /opt/splunk
+          sudo chown -R "$USER":"$USER" $SPLUNK_HOME
+          cp -r tests/integration/data/solnlib_demo $SPLUNK_HOME/etc/apps
+          cp -r solnlib $SPLUNK_HOME/etc/apps/solnlib_demo/bin/
+          mkdir -p $SPLUNK_HOME/etc/apps/Splunk_TA_test/default/
           ls $SPLUNK_HOME/etc/apps/solnlib_demo/bin/
-          echo -e "[user_info]\nUSERNAME=Admin\nPASSWORD=Chang3d"'!' | sudo tee -a /opt/splunk/etc/system/local/user-seed.conf
-          echo 'OPTIMISTIC_ABOUT_FILE_LOCKING=1' | sudo tee -a /opt/splunk/etc/splunk-launch.conf
-          sudo /opt/splunk/bin/splunk start --accept-license
-          sudo /opt/splunk/bin/splunk cmd python -m pip install solnlib
-          sudo /opt/splunk/bin/splunk set servername custom-servername -auth admin:Chang3d!
-          sudo /opt/splunk/bin/splunk restart
+          echo -e "[user_info]\nUSERNAME=Admin\nPASSWORD=Chang3d"'!' | tee -a $SPLUNK_HOME/etc/system/local/user-seed.conf
+          echo 'OPTIMISTIC_ABOUT_FILE_LOCKING=1' | tee -a $SPLUNK_HOME/etc/splunk-launch.conf
+          $SPLUNK_HOME/bin/splunk start --accept-license
+          $SPLUNK_HOME/bin/splunk cmd python -m pip install solnlib
+          $SPLUNK_HOME/bin/splunk set servername custom-servername -auth admin:Chang3d!
+          $SPLUNK_HOME/bin/splunk restart
           until curl -k -s -u admin:Chang3d! https://localhost:8089/services/server/info\?output_mode\=json | jq '.entry[0].content.kvStoreStatus' | grep -o "ready" ; do echo -n "Waiting for KVStore to become ready-" && sleep 5 ; done
         timeout-minutes: 5
       - name: Run tests
         run: |
           poetry install
-          SPLUNK_HOME=/opt/splunk/ poetry run pytest --junitxml=test-results/results.xml -v tests/integration
+          SPLUNK_HOME=/opt/splunk SPLUNK_DB=$SPLUNK_HOME/var/lib/splunk poetry run pytest --junitxml=test-results/results.xml -v tests/integration
       - uses: actions/upload-artifact@v4
         with:
           name: test-splunk-${{ matrix.splunk.version }}

.gitignore

@@ -1,11 +1,14 @@
 # IDE related files
 *.idea
 *.DS_Store*
-.venv/
+# ignore all virtual environments
+.venv*
 
 # Compiled files
 __pycache__
 *.pyc
 *.pyo
 
 .coverage
+*.log
+events.pickle

solnlib/server_info.py

@@ -16,13 +16,15 @@
 
 """This module contains Splunk server info related functionalities."""
 
+import os
 import json
 from typing import Any, Dict, Optional
 
+from splunk.rest import getWebCertFile, getWebKeyFile
 from splunklib import binding
-
 from solnlib import splunk_rest_client as rest_client
 from solnlib import utils
+from solnlib.splunkenv import get_splunkd_access_info
 
 __all__ = ["ServerInfo", "ServerInfoException"]
 
@@ -56,6 +58,28 @@ def __init__(
             port: The port number, default is None.
             context: Other configurations for Splunk rest client.
         """
+        is_localhost = False
+        if not all([scheme, host, port]) and os.environ.get("SPLUNK_HOME"):
+            scheme, host, port = get_splunkd_access_info()
+            is_localhost = (
+                host == "localhost" or host == "127.0.0.1" or host in ("::1", "[::1]")
+            )
+
+        if getWebCertFile() and getWebKeyFile():
+            context["cert_file"] = getWebCertFile()
+            context["key_file"] = getWebKeyFile()
+
+            if all([is_localhost, context.get("verify") is None]):
+                # NOTE: this is specifically for mTLS communication
+                # ONLY if scheme, host, port aren't provided AND user hasn't provided server certificate
+                # we set verify to off (similar to 'rest.simpleRequest' implementation)
+                context["verify"] = False
+
+        elif getWebCertFile() is not None:
+            context["cert_file"] = getWebCertFile()
+            if all([is_localhost, context.get("verify") is None]):
+                context["verify"] = False
+
         self._rest_client = rest_client.SplunkRestClient(
             session_key, "-", scheme=scheme, host=host, port=port, **context
         )

solnlib/splunk_rest_client.py

@@ -88,10 +88,13 @@ def _request_handler(context):
     verify = context.get("verify", False)
 
     if context.get("key_file") and context.get("cert_file"):
-        # cert = ('/path/client.cert', '/path/client.key')
-        cert = context["key_file"], context["cert_file"]
+        # cert: if tuple, ('cert', 'key') pair as per requests library
+        cert = context["cert_file"], context["key_file"]
     elif context.get("cert_file"):
         cert = context["cert_file"]
+    elif context.get("cert"):
+        # as the solnlib uses requests, we need to have a check for 'cert' key as well
+        cert = context["cert"]
     else:
         cert = None
 

tests/integration/conftest.py

@@ -0,0 +1,8 @@
+import os
+import sys
+
+# path manipulation get the 'splunk' library for the imports while running on GH Actions
+sys.path.append(
+    os.path.sep.join([os.environ["SPLUNK_HOME"], "lib", "python3.7", "site-packages"])
+)
+# TODO: 'python3.7' needs to be updated as and when Splunk has new folder for Python.

tests/unit/conftest.py

@@ -0,0 +1,6 @@
+import sys
+from unittest.mock import MagicMock
+
+# mock modules of 'splunk' library added 'splunk_rest_client'
+sys.modules["splunk"] = MagicMock()
+sys.modules["splunk.rest"] = MagicMock()

tests/unit/test_server_info.py

@@ -13,8 +13,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-import common
 import pytest
+import common
+from unittest.mock import patch, MagicMock
 from splunklib import binding
 
 from solnlib import server_info
@@ -72,3 +73,91 @@ def _mock_get(self, path_segment, owner=None, app=None, sharing=None, **query):
 
         si = server_info.ServerInfo(common.SESSION_KEY)
         assert si.is_captain_ready()
+
+    @patch("solnlib.server_info.os.environ", autospec=True, return_value="$SPLUNK_HOME")
+    @patch(
+        "solnlib.server_info.get_splunkd_access_info",
+        autospec=True,
+        return_value=("https", "127.0.0.1", "8089"),
+    )
+    @patch("solnlib.server_info.rest_client", autospec=True)
+    @patch("solnlib.server_info.getWebCertFile", return_value=None)
+    @patch("solnlib.server_info.getWebKeyFile", return_value=None)
+    def test_server_info_object_with_no_certs(
+        self, mock_web_key, mock_web_cert, mock_rest_client, mock_splunkd, mock_os_env
+    ):
+        mock_rest_client.SplunkRestClient = MagicMock()
+
+        server_info.ServerInfo(common.SESSION_KEY)
+
+        for call_arg in mock_rest_client.SplunkRestClient.call_args_list:
+            _, kwargs = call_arg
+            assert kwargs.get("cert_file") is None
+            assert kwargs.get("key_file") is None
+            assert kwargs.get("verify") is None
+
+    @patch("solnlib.server_info.os.environ", autospec=True, return_value="$SPLUNK_HOME")
+    @patch(
+        "solnlib.server_info.get_splunkd_access_info",
+        autospec=True,
+        return_value=("https", "127.0.0.1", "8089"),
+    )
+    @patch("solnlib.server_info.rest_client", autospec=True)
+    @patch("solnlib.server_info.getWebCertFile", return_value="/path/cert/pem")
+    @patch("solnlib.server_info.getWebKeyFile", return_value="/path/key/pem")
+    def test_server_info_object_with_both_certs(
+        self, mock_web_key, mock_web_cert, mock_rest_client, mock_splunkd, mock_os_env
+    ):
+        mock_rest_client.SplunkRestClient = MagicMock()
+
+        server_info.ServerInfo(common.SESSION_KEY)
+
+        for call_arg in mock_rest_client.SplunkRestClient.call_args_list:
+            _, kwargs = call_arg
+            assert kwargs.get("cert_file") == "/path/cert/pem"
+            assert kwargs.get("key_file") == "/path/key/pem"
+            assert kwargs.get("verify") is False
+
+    @patch("solnlib.server_info.os.environ", autospec=True, return_value="$SPLUNK_HOME")
+    @patch(
+        "solnlib.server_info.get_splunkd_access_info",
+        autospec=True,
+        return_value=("https", "127.0.0.1", "8089"),
+    )
+    @patch("solnlib.server_info.rest_client", autospec=True)
+    @patch("solnlib.server_info.getWebCertFile", return_value="/path/cert/pem")
+    @patch("solnlib.server_info.getWebKeyFile", return_value=None)
+    def test_server_info_object_with_cert_file(
+        self, mock_web_key, mock_web_cert, mock_rest_client, mock_splunkd, mock_os_env
+    ):
+        mock_rest_client.SplunkRestClient = MagicMock()
+
+        server_info.ServerInfo(common.SESSION_KEY)
+
+        for call_arg in mock_rest_client.SplunkRestClient.call_args_list:
+            _, kwargs = call_arg
+            assert kwargs.get("cert_file") == "/path/cert/pem"
+            assert kwargs.get("key_file") is None
+            assert kwargs.get("verify") is False
+
+    @patch("solnlib.server_info.os.environ", autospec=True, return_value="$SPLUNK_HOME")
+    @patch(
+        "solnlib.server_info.get_splunkd_access_info",
+        autospec=True,
+        return_value=("https", "127.0.0.1", "8089"),
+    )
+    @patch("solnlib.server_info.rest_client", autospec=True)
+    @patch("solnlib.server_info.getWebCertFile", return_value=None)
+    @patch("solnlib.server_info.getWebKeyFile", return_value="/path/key/pem")
+    def test_server_info_object_with_key_file(
+        self, mock_web_key, mock_web_cert, mock_rest_client, mock_splunkd, mock_os_env
+    ):
+        mock_rest_client.SplunkRestClient = MagicMock()
+
+        server_info.ServerInfo(common.SESSION_KEY)
+
+        for call_arg in mock_rest_client.SplunkRestClient.call_args_list:
+            _, kwargs = call_arg
+            assert kwargs.get("cert_file") is None
+            assert kwargs.get("key_file") is None
+            assert kwargs.get("verify") is None