feat: Warn on insecure tool invocation with authentication (#223)

* chore: Add unit test cases

* chore: Delint

* feat: Warn on insecure tool invocation with authentication

This change introduces a warning that is displayed immediately before a tool invocation if:
1. The invocation includes an authentication header.
2. The connection is being made over non-secure HTTP.

> [!IMPORTANT]
The purpose of this warning is to alert the user to the security risk of sending credentials over an unencrypted channel and to encourage the use of HTTPS.

* fix!: Warn about https only during tool initialization

* chore: Add unit test coverage

* chore: Delint

Author: anubhav756

packages/toolbox-core/src/toolbox_core/tool.py

@@ -16,6 +16,7 @@
 from inspect import Signature
 from types import MappingProxyType
 from typing import Any, Callable, Coroutine, Mapping, Optional, Sequence, Union
+from warnings import warn
 
 from aiohttp import ClientSession
 
@@ -118,6 +119,17 @@ def __init__(
         # map of client headers to their value/callable/coroutine
         self.__client_headers = client_headers
 
+        # ID tokens contain sensitive user information (claims). Transmitting
+        # these over HTTP exposes the data to interception and unauthorized
+        # access. Always use HTTPS to ensure secure communication and protect
+        # user privacy.
+        if (
+            required_authn_params or required_authz_tokens or client_headers
+        ) and not self.__url.startswith("https://"):
+            warn(
+                "Sending ID token over HTTP. User data may be exposed. Use HTTPS for secure communication."
+            )
+
     @property
     def _name(self) -> str:
         return self.__name__
@@ -327,7 +339,8 @@ def add_auth_token_getters(
             )
 
         return self.__copy(
-            # create a read-only map for updated getters, params and tokens that are still required
+            # create read-only values for updated getters, params and tokens
+            # that are still required
             auth_service_token_getters=MappingProxyType(new_getters),
             required_authn_params=MappingProxyType(new_req_authn_params),
             required_authz_tokens=tuple(new_req_authz_tokens),

packages/toolbox-core/tests/test_tool.py

@@ -16,6 +16,7 @@
 import inspect
 from typing import AsyncGenerator, Callable, Mapping
 from unittest.mock import AsyncMock, Mock
+from warnings import catch_warnings, simplefilter
 
 import pytest
 import pytest_asyncio
@@ -27,6 +28,7 @@
 from toolbox_core.tool import ToolboxTool, create_func_docstring, resolve_value
 
 TEST_BASE_URL = "http://toolbox.example.com"
+HTTPS_BASE_URL = "https://toolbox.example.com"
 TEST_TOOL_NAME = "sample_tool"
 
 
@@ -195,7 +197,7 @@ async def test_tool_creation_callable_and_run(
     Tests creating a ToolboxTool, checks callability, and simulates a run.
     """
     tool_name = TEST_TOOL_NAME
-    base_url = TEST_BASE_URL
+    base_url = HTTPS_BASE_URL
     invoke_url = f"{base_url}/api/tool/{tool_name}/invoke"
 
     input_args = {"message": "hello world", "count": 5}
@@ -246,7 +248,7 @@ async def test_tool_run_with_pydantic_validation_error(
     due to Pydantic validation *before* making an HTTP request.
     """
     tool_name = TEST_TOOL_NAME
-    base_url = TEST_BASE_URL
+    base_url = HTTPS_BASE_URL
     invoke_url = f"{base_url}/api/tool/{tool_name}/invoke"
 
     with aioresponses() as m:
@@ -340,18 +342,25 @@ async def test_resolve_value_async_callable():
 
 def test_tool_init_basic(http_session, sample_tool_params, sample_tool_description):
     """Tests basic tool initialization without headers or auth."""
-    tool_instance = ToolboxTool(
-        session=http_session,
-        base_url=TEST_BASE_URL,
-        name=TEST_TOOL_NAME,
-        description=sample_tool_description,
-        params=sample_tool_params,
-        required_authn_params={},
-        required_authz_tokens=[],
-        auth_service_token_getters={},
-        bound_params={},
-        client_headers={},
-    )
+    with catch_warnings(record=True) as record:
+        simplefilter("always")
+
+        tool_instance = ToolboxTool(
+            session=http_session,
+            base_url=HTTPS_BASE_URL,
+            name=TEST_TOOL_NAME,
+            description=sample_tool_description,
+            params=sample_tool_params,
+            required_authn_params={},
+            required_authz_tokens=[],
+            auth_service_token_getters={},
+            bound_params={},
+            client_headers={},
+        )
+    assert (
+        len(record) == 0
+    ), f"ToolboxTool instantiation unexpectedly warned: {[f'{w.category.__name__}: {w.message}' for w in record]}"
+
     assert tool_instance.__name__ == TEST_TOOL_NAME
     assert inspect.iscoroutinefunction(tool_instance.__call__)
     assert "message" in tool_instance.__signature__.parameters
@@ -367,7 +376,7 @@ def test_tool_init_with_client_headers(
     """Tests tool initialization *with* client headers."""
     tool_instance = ToolboxTool(
         session=http_session,
-        base_url=TEST_BASE_URL,
+        base_url=HTTPS_BASE_URL,
         name=TEST_TOOL_NAME,
         description=sample_tool_description,
         params=sample_tool_params,
@@ -395,7 +404,7 @@ def test_tool_init_header_auth_conflict(
     ):
         ToolboxTool(
             session=http_session,
-            base_url=TEST_BASE_URL,
+            base_url=HTTPS_BASE_URL,
             name="auth_conflict_tool",
             description=sample_tool_description,
             params=sample_tool_auth_params,
@@ -418,7 +427,7 @@ def test_tool_add_auth_token_getters_conflict_with_existing_client_header(
     """
     tool_instance = ToolboxTool(
         session=http_session,
-        base_url=TEST_BASE_URL,
+        base_url=HTTPS_BASE_URL,
         name="tool_with_client_header",
         description=sample_tool_description,
         params=sample_tool_params,
@@ -454,7 +463,7 @@ def test_add_auth_token_getters_unused_token(
     """
     tool_instance = ToolboxTool(
         session=http_session,
-        base_url=TEST_BASE_URL,
+        base_url=HTTPS_BASE_URL,
         name=TEST_TOOL_NAME,
         description=sample_tool_description,
         params=sample_tool_params,
@@ -469,3 +478,131 @@ def test_add_auth_token_getters_unused_token(
 
     with pytest.raises(ValueError, match=expected_error_message):
         tool_instance.add_auth_token_getters(unused_auth_getters)
+
+
+# --- Test for the HTTP Warning ---
+@pytest.mark.parametrize(
+    "trigger_condition_params",
+    [
+        {"client_headers": {"X-Some-Header": "value"}},
+        {"required_authn_params": {"param1": ["auth-service1"]}},
+        {"required_authz_tokens": ["auth-service2"]},
+        {
+            "client_headers": {"X-Some-Header": "value"},
+            "required_authn_params": {"param1": ["auth-service1"]},
+        },
+        {
+            "client_headers": {"X-Some-Header": "value"},
+            "required_authz_tokens": ["auth-service2"],
+        },
+        {
+            "required_authn_params": {"param1": ["auth-service1"]},
+            "required_authz_tokens": ["auth-service2"],
+        },
+        {
+            "client_headers": {"X-Some-Header": "value"},
+            "required_authn_params": {"param1": ["auth-service1"]},
+            "required_authz_tokens": ["auth-service2"],
+        },
+    ],
+    ids=[
+        "client_headers_only",
+        "authn_params_only",
+        "authz_tokens_only",
+        "headers_and_authn",
+        "headers_and_authz",
+        "authn_and_authz",
+        "all_three_conditions",
+    ],
+)
+def test_tool_init_http_warning_when_sensitive_info_over_http(
+    http_session: ClientSession,
+    sample_tool_params: list[ParameterSchema],
+    sample_tool_description: str,
+    trigger_condition_params: dict,
+):
+    """
+    Tests that a UserWarning is issued if client headers, auth params, or
+    auth tokens are present and the base_url is HTTP.
+    """
+    expected_warning_message = (
+        "Sending ID token over HTTP. User data may be exposed. "
+        "Use HTTPS for secure communication."
+    )
+
+    init_kwargs = {
+        "session": http_session,
+        "base_url": TEST_BASE_URL,
+        "name": "http_warning_tool",
+        "description": sample_tool_description,
+        "params": sample_tool_params,
+        "required_authn_params": {},
+        "required_authz_tokens": [],
+        "auth_service_token_getters": {},
+        "bound_params": {},
+        "client_headers": {},
+    }
+    # Apply the specific conditions for this parametrized test
+    init_kwargs.update(trigger_condition_params)
+
+    with pytest.warns(UserWarning, match=expected_warning_message):
+        ToolboxTool(**init_kwargs)
+
+
+def test_tool_init_no_http_warning_if_https(
+    http_session: ClientSession,
+    sample_tool_params: list[ParameterSchema],
+    sample_tool_description: str,
+    static_client_header: dict,
+):
+    """
+    Tests that NO UserWarning is issued if client headers are present but
+    the base_url is HTTPS.
+    """
+    with catch_warnings(record=True) as record:
+        simplefilter("always")
+
+        ToolboxTool(
+            session=http_session,
+            base_url=HTTPS_BASE_URL,
+            name="https_tool",
+            description=sample_tool_description,
+            params=sample_tool_params,
+            required_authn_params={},
+            required_authz_tokens=[],
+            auth_service_token_getters={},
+            bound_params={},
+            client_headers=static_client_header,
+        )
+    assert (
+        len(record) == 0
+    ), f"Expected no warnings, but got: {[f'{w.category.__name__}: {w.message}' for w in record]}"
+
+
+def test_tool_init_no_http_warning_if_no_sensitive_info_on_http(
+    http_session: ClientSession,
+    sample_tool_params: list[ParameterSchema],
+    sample_tool_description: str,
+):
+    """
+    Tests that NO UserWarning is issued if the URL is HTTP but there are
+    no client headers, auth params, or auth tokens.
+    """
+    with catch_warnings(record=True) as record:
+        simplefilter("always")
+
+        ToolboxTool(
+            session=http_session,
+            base_url=TEST_BASE_URL,
+            name="http_tool_no_sensitive",
+            description=sample_tool_description,
+            params=sample_tool_params,
+            required_authn_params={},
+            required_authz_tokens=[],
+            auth_service_token_getters={},
+            bound_params={},
+            client_headers={},
+        )
+    assert (
+        len(record) == 0
+    ), f"Expected no warnings, but got: {[f'{w.category.__name__}: {w.message}' for w in record]}"