feat Added prefix to env vars and fixed convention of return outputs

Signed-off-by: S3B4SZ17 <[email protected]>

Author: S3B4SZ17

.github/workflows/test.yaml

@@ -45,57 +45,3 @@ jobs:
 
       - name: Run Unit Tests
         run: make test
-
-  check_version:
-    name: Check Version
-    runs-on: ubuntu-latest
-    needs: test
-    permissions:
-      contents: write # required for creating a tag
-    steps:
-      - name: Check out repository
-        uses: actions/checkout@v4
-        with:
-          ref: ${{ github.sha }} # required for better experience using pre-releases
-          fetch-depth: '0' # Required due to the way Git works, without it this action won't be able to find any or the correct tags
-
-      - name: Extract current version
-        id: pyproject_version
-        run: |
-          TAG=v$(grep 'version =' pyproject.toml | sed -e 's/version = "\(.*\)"/\1/')
-          echo "TAG=$TAG" >> "$GITHUB_OUTPUT"
-
-      - name: Get branch ref name
-        id: branch_ref
-        run: |
-          BRANCH_NAME=${{ github.base_ref || github.ref_name }}
-          echo "$BRANCH_NAME"
-          echo "BRANCH_NAME=$BRANCH_NAME" >> "$GITHUB_OUTPUT"
-
-      - name: Get tag version
-        id: semantic_release
-        uses: anothrNick/[email protected]
-        env:
-          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-          DEFAULT_BUMP: "patch"
-          TAG_CONTEXT: 'repo'
-          WITH_V: true
-          DRY_RUN: true
-
-      - name: Compare versions
-        run: |
-          echo "Current version: ${{ steps.pyproject_version.outputs.TAG }}"
-          echo "New version: ${{ steps.semantic_release.outputs.tag }}"
-          if [ "${{ steps.pyproject_version.outputs.TAG }}" != "${{ steps.semantic_release.outputs.tag }}" ]; then
-            echo "### Version mismatch detected! :warning:
-            Current pyproject version: ${{ steps.pyproject_version.outputs.TAG }}
-            New Tag version: **${{ steps.semantic_release.outputs.tag }}**
-            Current Tag: ${{ steps.semantic_release.outputs.old_tag }}
-            Please update the version in pyproject.toml." >> $GITHUB_STEP_SUMMARY
-            exit 1
-          else
-            echo "### Version match confirmed! :rocket:
-            Current pyproject version: ${{ steps.pyproject_version.outputs.TAG }}
-            New Tag version: **${{ steps.semantic_release.outputs.tag }}**
-            The version is up-to-date." >> $GITHUB_STEP_SUMMARY
-          fi

main.py

@@ -27,8 +27,6 @@
 log = logging.getLogger(__name__)
 
 
-
-
 def handle_signals():
     def signal_handler(sig, frame):
         log.info(f"Received signal {sig}, shutting down...")

pyproject.toml

@@ -10,7 +10,7 @@ dependencies = [
     "pyyaml==6.0.2",
     "sqlalchemy==2.0.36",
     "sqlmodel==0.0.22",
-    "sysdig-sdk-python @ git+https://github.com/sysdiglabs/sysdig-sdk-python@597285143188019cd0e86fde43f94b1139f5441d",
+    "sysdig-sdk-python @ git+https://github.com/sysdiglabs/sysdig-sdk-python@852ee2ccad12a8b445dd4732e7f3bd44d78a37f7",
     "dask==2025.4.1",
     "oauthlib==3.2.2",
     "fastapi==0.116.1",

tests/conftest.py

@@ -5,7 +5,12 @@
 import json
 import pytest
 import os
-from unittest.mock import patch
+from utils.app_config import AppConfig
+from unittest.mock import MagicMock, create_autospec, patch
+from fastmcp.server.context import Context
+from sysdig_client import SecureEventsApi, ApiClient, InventoryApi, VulnerabilityManagementApi
+from utils.sysdig.legacy_sysdig_api import LegacySysdigApi
+from fastmcp.server import FastMCP
 
 
 def util_load_json(path):
@@ -26,7 +31,7 @@ def mock_success_response():
     Fixture to mock the urllib3.PoolManager.request method
 
     Yields:
-                MagicMock: A mocked request object that simulates a successful HTTP response.
+        MagicMock: A mocked request object that simulates a successful HTTP response.
     """
     with patch("urllib3.PoolManager.request") as mock_request:
         mock_resp = patch("urllib3.response.HTTPResponse").start()
@@ -42,5 +47,41 @@ def mock_creds():
     """
     Fixture to set up mocked credentials.
     """
-    os.environ["SYSDIG_SECURE_TOKEN"] = "mocked_token"
-    os.environ["SYSDIG_HOST"] = "https://us2.app.sysdig.com"
+    os.environ["SYSDIG_MCP_SECURE_TOKEN"] = "mocked_token"
+    os.environ["SYSDIG_MCP_HOST"] = "https://us2.app.sysdig.com"
+
+
+def mock_app_config() -> AppConfig:
+    """
+    Utility function to create a mocked AppConfig instance.
+    Returns:
+        AppConfig: A mocked AppConfig instance.
+    """
+    mock_cfg = create_autospec(AppConfig, instance=True)
+
+    mock_cfg.sysdig_endpoint.return_value = "https://us2.app.sysdig.com"
+    mock_cfg.transport.return_value = "stdio"
+    mock_cfg.log_level.return_value = "DEBUG"
+    mock_cfg.port.return_value = 8080
+
+    return mock_cfg
+
+
+@pytest.fixture
+def mock_context() -> Context:
+    """
+    Utility function to create a mocked FastMCP context.
+    Returns:
+        Context: A mocked FastMCP context.
+    """
+
+    ctx = Context(MagicMock(spec=FastMCP))
+
+    api_instances = {
+        "secure_events": SecureEventsApi(ApiClient()),
+        "vulnerability_management": VulnerabilityManagementApi(ApiClient()),
+        "inventory": InventoryApi(ApiClient()),
+        "legacy_sysdig_api": LegacySysdigApi(ApiClient()),
+    }
+    ctx.set_state("api_instances", api_instances)
+    return ctx

tests/events_feed_test.py

@@ -2,15 +2,13 @@
 Events Feed Test Module
 """
 
+import os
 from http import HTTPStatus
 from tools.events_feed.tool import EventsFeedTools
-from utils.app_config import AppConfig
-from .conftest import util_load_json
-from unittest.mock import MagicMock, AsyncMock, create_autospec
-from sysdig_client.api import SecureEventsApi
-import os
+from unittest.mock import MagicMock, AsyncMock
 from fastmcp.server.context import Context
 from fastmcp.server import FastMCP
+from .conftest import util_load_json, mock_app_config
 
 # Get the absolute path of the current module file
 module_path = os.path.abspath(__file__)
@@ -20,18 +18,10 @@
 
 EVENT_INFO_RESPONSE = util_load_json(f"{module_directory}/test_data/events_feed/event_info_response.json")
 
+ctx = Context(MagicMock(spec=FastMCP))
 
-def mock_app_config() -> AppConfig:
-    mock_cfg = create_autospec(AppConfig, instance=True)
-
-    mock_cfg.sysdig_endpoint.return_value = "https://us2.app.sysdig.com"
-    mock_cfg.transport.return_value = "stdio"
-    mock_cfg.log_level.return_value = "DEBUG"
-    mock_cfg.port.return_value = 8080
-
-    return mock_cfg
 
-def test_get_event_info(mock_success_response: MagicMock | AsyncMock, mock_creds) -> None:
+def test_get_event_info(mock_success_response: MagicMock | AsyncMock, mock_creds, mock_context: Context) -> None:
     """Test the get_event_info tool method.
     Args:
         mock_success_response (MagicMock | AsyncMock): Mocked response object.
@@ -43,22 +33,8 @@ def test_get_event_info(mock_success_response: MagicMock | AsyncMock, mock_creds
 
     tools_client = EventsFeedTools(app_config=mock_app_config())
 
-    ctx = Context(FastMCP())
-
-    # Seed FastMCP Context state with mocked API instances expected by the tools
-    secure_events_api = MagicMock(spec=SecureEventsApi)
-    # The tool returns whatever the SDK method returns; make it be our mocked HTTP response
-    secure_events_api.get_event_v1_without_preload_content.return_value = mock_success_response.return_value
-
-    api_instances = {
-        "secure_events": secure_events_api,
-        # Not used by this test, but present in real runtime; keep as empty mock to avoid KeyErrors elsewhere
-        "legacy_sysdig_api": MagicMock(),
-    }
-    ctx.set_state("api_instances", api_instances)
-
     # Pass the mocked Context object
-    result: dict = tools_client.tool_get_event_info(ctx=ctx, event_id="12345")
+    result: dict = tools_client.tool_get_event_info(ctx=mock_context, event_id="12345")
     results: dict = result["results"]
 
     assert result.get("status_code") == HTTPStatus.OK

tools/cli_scanner/tool.py

@@ -11,14 +11,16 @@
 
 from utils.app_config import AppConfig
 
+TMP_OUTPUT_FILE = "/tmp/sysdig_cli_scanner_output.json"
+
+
 class CLIScannerTool:
     """
     A class to encapsulate the tools for interacting with the Sysdig CLI Scanner.
     """
 
     def __init__(self, app_config: AppConfig):
         self.app_config = app_config
-        logging.basicConfig(format="%(asctime)s-%(process)d-%(levelname)s- %(message)s", level=app_config.log_level())
         self.log = logging.getLogger(__name__)
         self.cmd: str = "sysdig-cli-scanner"
         self.default_args: list = [
@@ -146,7 +148,7 @@ def run_sysdig_cli_scanner(
 
         try:
             # Run the command
-            with open("sysdig_cli_scanner_output.json", "w") as output_file:
+            with open(TMP_OUTPUT_FILE, "w") as output_file:
                 result = subprocess.run(cmd, text=True, check=True, stdout=output_file, stderr=subprocess.PIPE)
                 output_result = output_file.read()
                 output_file.close()
@@ -168,15 +170,15 @@ def run_sysdig_cli_scanner(
                 }
                 raise Exception(result)
             else:
-                with open("sysdig_cli_scanner_output.json", "r") as output_file:
+                with open(TMP_OUTPUT_FILE, "r") as output_file:
                     output_result = output_file.read()
                 result: dict = {
                     "exit_code": e.returncode,
                     "stdout": e.stdout,
                     "output": output_result,
                     "exit_codes_explained": self.exit_code_explained,
                 }
-                os.remove("sysdig_cli_scanner_output.json")
+                os.remove(TMP_OUTPUT_FILE)
                 return result
         # Handle any other exceptions that may occur and exit codes 2 and 3
         except Exception as e:

tools/events_feed/tool.py

@@ -111,10 +111,10 @@ def tool_list_runtime_events(
         Returns:
             dict: A dictionary containing the results of the runtime events query, including pagination information.
         """
+        start_time = time.time()
         api_instances: dict = ctx.get_state("api_instances")
         secure_events_api: SecureEventsApi = api_instances.get("secure_events")
 
-        start_time = time.time()
         # Compute time window
         now_ns = time.time_ns()
         from_ts = now_ns - scope_hours * 3600 * 1_000_000_000
@@ -160,10 +160,10 @@ def tool_get_event_process_tree(self, ctx: Context, event_id: str) -> dict:
             ToolError: If there is an error constructing or processing the response.
         """
         try:
+            start_time = time.time()
             api_instances: dict = ctx.get_state("api_instances")
             legacy_api_client: LegacySysdigApi = api_instances.get("legacy_sysdig_api")
 
-            start_time = time.time()
             # Get process tree branches
             branches = legacy_api_client.request_process_tree_branches(event_id)
             # Get process tree
@@ -194,7 +194,7 @@ def tool_get_event_process_tree(self, ctx: Context, event_id: str) -> dict:
                     "metadata": {
                         "execution_time_ms": (time.time() - start_time) * 1000,
                         "timestamp": datetime.datetime.now(datetime.UTC).isoformat().replace("+00:00", "Z"),
-                        "note": "Process tree not available for this event"
+                        "note": "Process tree not available for this event",
                     },
                 }
             else:

tools/inventory/tool.py

@@ -19,6 +19,7 @@ class InventoryTools:
     A class to encapsulate the tools for interacting with the Sysdig Secure Inventory API.
     This class provides methods to list resources and retrieve a single resource by its hash.
     """
+
     def __init__(self, app_config: AppConfig):
         self.app_config = app_config
         # Configure logging
@@ -35,7 +36,7 @@ def tool_list_resources(
                     Sysdig Secure query filter expression to filter inventory resources.
 
                     Use the resource://filter-query-language to get the expected filter expression format.
-               
+
                     List of supported fields:
                     - accountName
                     - accountId
@@ -138,11 +139,10 @@ def tool_list_resources(
             Or a dict containing an error message if the call fails.
         """
         try:
+            start_time = time.time()
             api_instances: dict = ctx.get_state("api_instances")
             inventory_api: InventoryApi = api_instances.get("inventory")
 
-            start_time = time.time()
-
             api_response = inventory_api.get_resources_without_preload_content(
                 filter=filter_exp, page_number=page_number, page_size=page_size, with_enriched_containers=with_enrich_containers
             )
@@ -172,11 +172,10 @@ def tool_get_resource(
             dict: A dictionary containing the details of the requested inventory resource.
         """
         try:
+            start_time = time.time()
             api_instances: dict = ctx.get_state("api_instances")
             inventory_api: InventoryApi = api_instances.get("inventory")
 
-            start_time = time.time()
-
             api_response = inventory_api.get_resource_without_preload_content(hash=resource_hash)
             execution_time = (time.time() - start_time) * 1000
 

tools/sysdig_sage/tool.py

@@ -20,6 +20,7 @@ class SageTools:
     This class provides methods to generate SysQL queries based on natural
     language questions and execute them against the Sysdig API.
     """
+
     def __init__(self, app_config: AppConfig):
         self.app_config = app_config
         self.log = logging.getLogger(__name__)