From 2014f99ff30d9d5ae13bbb0d3de6c2012fde8391 Mon Sep 17 00:00:00 2001 From: Anubhav Dhawan Date: Thu, 20 Aug 2026 11:16:48 +0530 Subject: [PATCH] feat(core): tool-level secure parameter binding, fast-fail and validation --- .../src/toolbox_core/sync_tool.py | 55 ++++ .../toolbox-core/src/toolbox_core/tool.py | 150 ++++++++- .../toolbox-core/src/toolbox_core/utils.py | 27 +- packages/toolbox-core/tests/test_sync_tool.py | 58 +++- packages/toolbox-core/tests/test_tool.py | 303 ++++++++++++++++++ packages/toolbox-core/tests/test_utils.py | 70 ++++ 6 files changed, 646 insertions(+), 17 deletions(-) diff --git a/packages/toolbox-core/src/toolbox_core/sync_tool.py b/packages/toolbox-core/src/toolbox_core/sync_tool.py index 61976a2e4..d4a832bcf 100644 --- a/packages/toolbox-core/src/toolbox_core/sync_tool.py +++ b/packages/toolbox-core/src/toolbox_core/sync_tool.py @@ -101,12 +101,22 @@ def _description(self) -> str: def _params(self) -> Sequence[ParameterSchema]: return self.__async_tool._params + @property + def _secure_params(self) -> Sequence[ParameterSchema]: + return self.__async_tool._secure_params + @property def _bound_params( self, ) -> Mapping[str, Union[Callable[[], Any], Callable[[], Awaitable[Any]], Any]]: return self.__async_tool._bound_params + @property + def _bound_secure_params( + self, + ) -> Mapping[str, Union[Callable[[], Any], Callable[[], Awaitable[Any]], Any]]: + return self.__async_tool._bound_secure_params + @property def _required_authn_params(self) -> Mapping[str, list[str]]: return self.__async_tool._required_authn_params @@ -241,6 +251,51 @@ def bind_param( """ return self.bind_params({param_name: param_value}) + def bind_secure_params( + self, + bound_secure_params: Mapping[ + str, Union[Callable[[], Any], Callable[[], Awaitable[Any]], Any] + ], + ) -> "ToolboxSyncTool": + """ + Binds secure parameters to values or callables that produce values. + + Args: + bound_secure_params: A mapping of secure parameter names to values or + callables that produce values. + + Returns: + A new ToolboxSyncTool instance with the specified secure parameters bound. + + Raises: + ValueError: If a parameter is already bound or is not defined as a + secure parameter by the tool's definition. + """ + new_async_tool = self.__async_tool.bind_secure_params(bound_secure_params) + return ToolboxSyncTool(new_async_tool, self.__loop, self.__thread) + + def bind_secure_param( + self, + param_name: str, + param_value: Union[Callable[[], Any], Callable[[], Awaitable[Any]], Any], + ) -> "ToolboxSyncTool": + """ + Binds a secure parameter to the value or callable that produces the value. + + Args: + param_name: The name of the secure parameter to bind. + param_value: The value of the bound secure parameter, or a callable that + returns the value. + + Returns: + A new ToolboxSyncTool instance with the specified secure parameter bound. + + Raises: + ValueError: If the parameter is already bound or is not defined as a + secure parameter by the tool's definition. + """ + return self.bind_secure_params({param_name: param_value}) + def add_telemetry_attributes( self, telemetry_attributes: TelemetryAttributes ) -> "ToolboxSyncTool": diff --git a/packages/toolbox-core/src/toolbox_core/tool.py b/packages/toolbox-core/src/toolbox_core/tool.py index 89212325e..245be0fe7 100644 --- a/packages/toolbox-core/src/toolbox_core/tool.py +++ b/packages/toolbox-core/src/toolbox_core/tool.py @@ -59,6 +59,10 @@ def __init__( client_headers: Mapping[ str, Union[Callable[[], str], Callable[[], Awaitable[str]], str] ], + secure_params: Sequence[ParameterSchema] = (), + bound_secure_params: Mapping[ + str, Union[Callable[[], Any], Callable[[], Awaitable[Any]], Any] + ] = {}, ): """ Initializes a callable that will trigger the tool invocation through the @@ -78,11 +82,16 @@ def __init__( bound_params: A mapping of parameter names to bind to specific values or callables that are called to produce values as needed. client_headers: Client specific headers bound to the tool. + secure_params: The secure parameters of the tool. + bound_secure_params: A mapping of secure parameter names to bind to + specific values or callables. """ # used to invoke the toolbox API self.__transport = transport self.__description = description self.__params = params + self.__secure_params = secure_params + self.__bound_secure_params = bound_secure_params self.__pydantic_model = params_to_pydantic_model(name, self.__params) # Separate parameters into those without a default and those with a @@ -133,12 +142,22 @@ def _description(self) -> str: def _params(self) -> Sequence[ParameterSchema]: return copy.deepcopy(self.__params) + @property + def _secure_params(self) -> Sequence[ParameterSchema]: + return copy.deepcopy(self.__secure_params) + @property def _bound_params( self, ) -> Mapping[str, Union[Callable[[], Any], Callable[[], Awaitable[Any]], Any]]: return MappingProxyType(self.__bound_parameters) + @property + def _bound_secure_params( + self, + ) -> Mapping[str, Union[Callable[[], Any], Callable[[], Awaitable[Any]], Any]]: + return MappingProxyType(self.__bound_secure_params) + @property def _required_authn_params(self) -> Mapping[str, list[str]]: return MappingProxyType(self.__required_authn_params) @@ -177,6 +196,10 @@ def __copy( Mapping[str, Union[Callable[[], str], Callable[[], Awaitable[str]], str]] ] = None, telemetry_attributes: Optional[TelemetryAttributes] = None, + secure_params: Optional[Sequence[ParameterSchema]] = None, + bound_secure_params: Optional[ + Mapping[str, Union[Callable[[], Any], Callable[[], Awaitable[Any]], Any]] + ] = None, ) -> "ToolboxTool": """ Creates a copy of the ToolboxTool, overriding specific fields. @@ -197,6 +220,9 @@ def __copy( client_headers: Client specific headers bound to the tool. telemetry_attributes: Telemetry attributes for the derived tool. Set directly on the new instance (not exposed via __init__). + secure_params: The secure parameters of the tool. + bound_secure_params: A mapping of secure parameter names to bind to + specific values or callables. """ check = lambda val, default: val if val is not None else default new_tool = ToolboxTool( @@ -215,6 +241,8 @@ def __copy( ), bound_params=check(bound_params, self.__bound_parameters), client_headers=check(client_headers, self.__client_headers), + secure_params=check(secure_params, self.__secure_params), + bound_secure_params=check(bound_secure_params, self.__bound_secure_params), ) new_tool.__telemetry_attributes = check( telemetry_attributes, self.__telemetry_attributes @@ -255,6 +283,19 @@ async def __call__(self, *args: Any, **kwargs: Any) -> str: f": {','.join(req_auth_services)}" ) + # validate missing required secure parameters + missing_secure = [ + p.name + for p in self.__secure_params + if p.required + and p.default is None + and p.name not in self.__bound_secure_params + ] + if missing_secure: + raise ValueError( + f"Missing required secure parameter(s) {missing_secure} for tool '{self.__name__}'" + ) + # validate inputs to this call using the signature all_args = self.__signature__.bind(*args, **kwargs) @@ -274,6 +315,13 @@ async def __call__(self, *args: Any, **kwargs: Any) -> str: # error if it receives a None value, which it cannot convert. payload = OrderedDict({k: v for k, v in payload.items() if v is not None}) + # resolve secure bound parameters + secure_payload = {} + for param, value in self.__bound_secure_params.items(): + resolved = await resolve_value(value) + if resolved is not None: + secure_payload[param] = resolved + # create headers for auth services headers = {} for client_header_name, client_header_val in self.__client_headers.items(): @@ -287,17 +335,17 @@ async def __call__(self, *args: Any, **kwargs: Any) -> str: warn_if_http_and_headers(self.__transport.base_url, headers) + kwargs_to_pass: dict[str, Any] = {} if self.__telemetry_attributes is not None: - return await self.__transport.tool_invoke( - self.__name__, - payload, - headers, - telemetry_attributes=self.__telemetry_attributes, - ) + kwargs_to_pass["telemetry_attributes"] = self.__telemetry_attributes + if secure_payload: + kwargs_to_pass["secure_arguments"] = secure_payload + return await self.__transport.tool_invoke( self.__name__, payload, headers, + **kwargs_to_pass, ) def add_telemetry_attributes( @@ -431,17 +479,25 @@ def bind_params( A new ToolboxTool instance with the specified parameters bound. Raises: - ValueError: If a parameter is already bound or is not defined by the - tool's definition. + ValueError: If a parameter is already bound, is not defined by the + tool's definition, or is a secure parameter. """ param_names = set(p.name for p in self.__params) + secure_param_names = set(p.name for p in self.__secure_params) | set( + self.__bound_secure_params.keys() + ) for name in bound_params.keys(): if name in self.__bound_parameters: raise ValueError( f"cannot re-bind parameter: parameter '{name}' is already bound" ) + if name in secure_param_names: + raise ValueError( + f"parameter '{name}' is a secure parameter; use bind_secure_param/bind_secure_params instead" + ) + if name not in param_names: raise ValueError( f"unable to bind parameters: no parameter named {name}" @@ -476,8 +532,82 @@ def bind_param( A new ToolboxTool instance with the specified parameter bound. Raises: - ValueError: If the parameter is already bound or is not defined by - the tool's definition. + ValueError: If the parameter is already bound, is not defined by + the tool's definition, or is a secure parameter. """ return self.bind_params({param_name: param_value}) + + def bind_secure_params( + self, + bound_secure_params: Mapping[ + str, Union[Callable[[], Any], Callable[[], Awaitable[Any]], Any] + ], + ) -> "ToolboxTool": + """ + Binds secure parameters to values or callables that produce values. + + Args: + bound_secure_params: A mapping of secure parameter names to values or + callables that produce values. + + Returns: + A new ToolboxTool instance with the specified secure parameters bound. + + Raises: + ValueError: If a secure parameter is already bound, is a regular parameter, + or is not defined by the tool's secure parameter definition. + """ + secure_param_names = set(p.name for p in self.__secure_params) + regular_param_names = set(p.name for p in self.__params) | set( + self.__bound_parameters.keys() + ) + for name in bound_secure_params.keys(): + if name in self.__bound_secure_params: + raise ValueError( + f"cannot re-bind secure parameter: secure parameter '{name}' is already bound" + ) + + if name in regular_param_names: + raise ValueError( + f"parameter '{name}' is a regular parameter; use bind_param/bind_params instead" + ) + + if name not in secure_param_names: + raise ValueError( + f"unable to bind secure parameters: no secure parameter named {name}" + ) + + new_secure_params = [] + for p in self.__secure_params: + if p.name not in bound_secure_params: + new_secure_params.append(p) + all_bound_secure_params = dict(self.__bound_secure_params) + all_bound_secure_params.update(bound_secure_params) + + return self.__copy( + secure_params=new_secure_params, + bound_secure_params=MappingProxyType(all_bound_secure_params), + ) + + def bind_secure_param( + self, + param_name: str, + param_value: Union[Callable[[], Any], Callable[[], Awaitable[Any]], Any], + ) -> "ToolboxTool": + """ + Binds a secure parameter to the value or callable that produces the value. + + Args: + param_name: The name of the bound secure parameter. + param_value: The value of the bound secure parameter, or a callable that + returns the value. + + Returns: + A new ToolboxTool instance with the specified secure parameter bound. + + Raises: + ValueError: If the secure parameter is already bound, is a regular parameter, + or is not defined by the tool's definition. + """ + return self.bind_secure_params({param_name: param_value}) diff --git a/packages/toolbox-core/src/toolbox_core/utils.py b/packages/toolbox-core/src/toolbox_core/utils.py index 92660c84b..424115cb1 100644 --- a/packages/toolbox-core/src/toolbox_core/utils.py +++ b/packages/toolbox-core/src/toolbox_core/utils.py @@ -21,6 +21,7 @@ Callable, Iterable, Mapping, + Optional, Sequence, Type, Union, @@ -177,32 +178,46 @@ def validate_unused_requirements( name: str, is_toolset: bool = False, target_type: str | None = None, + provided_secure_keys: Optional[set[str]] = None, + used_secure_keys: Optional[set[str]] = None, ) -> None: """ - Validates that no provided authentication tokens or bound parameters went unused. + Validates that no provided authentication tokens, bound parameters, or secure parameters went unused. Raises a ValueError if any unused requirements are found, formatted appropriately for either a single tool or a full toolset. """ unused_auth = provided_auth_keys - used_auth_keys unused_bound = provided_bound_keys - used_bound_keys + unused_secure = (provided_secure_keys or set()) - (used_secure_keys or set()) - if unused_auth or unused_bound: + if unused_auth or unused_bound or unused_secure: error_messages = [] if unused_auth: if is_toolset: error_messages.append( - f"unused auth tokens could not be applied to any tool: {', '.join(unused_auth)}" + f"unused auth tokens could not be applied to any tool: {', '.join(sorted(unused_auth))}" ) else: - error_messages.append(f"unused auth tokens: {', '.join(unused_auth)}") + error_messages.append( + f"unused auth tokens: {', '.join(sorted(unused_auth))}" + ) if unused_bound: if is_toolset: error_messages.append( - f"unused bound parameters could not be applied to any tool: {', '.join(unused_bound)}" + f"unused bound parameters could not be applied to any tool: {', '.join(sorted(unused_bound))}" + ) + else: + error_messages.append( + f"unused bound parameters: {', '.join(sorted(unused_bound))}" + ) + if unused_secure: + if is_toolset: + error_messages.append( + f"unused secure parameters could not be applied to any tool: {', '.join(sorted(unused_secure))}" ) else: error_messages.append( - f"unused bound parameters: {', '.join(unused_bound)}" + f"unused secure parameters: {', '.join(sorted(unused_secure))}" ) final_target_type = ( diff --git a/packages/toolbox-core/tests/test_sync_tool.py b/packages/toolbox-core/tests/test_sync_tool.py index 7db4f94e1..7553571b4 100644 --- a/packages/toolbox-core/tests/test_sync_tool.py +++ b/packages/toolbox-core/tests/test_sync_tool.py @@ -21,7 +21,7 @@ import pytest -from toolbox_core.protocol import TelemetryAttributes +from toolbox_core.protocol import ParameterSchema, TelemetryAttributes from toolbox_core.sync_tool import ToolboxSyncTool from toolbox_core.tool import ToolboxTool @@ -46,6 +46,7 @@ def mock_async_tool() -> MagicMock: ToolboxTool, instance=True ) tool.bind_params.return_value = create_autospec(ToolboxTool, instance=True) + tool.bind_secure_params.return_value = create_autospec(ToolboxTool, instance=True) tool.add_telemetry_attributes.return_value = create_autospec( ToolboxTool, instance=True ) @@ -363,3 +364,58 @@ def test_toolbox_sync_tool_add_telemetry_attributes( assert new_sync_tool._ToolboxSyncTool__async_tool is new_mock_async_tool assert new_sync_tool._ToolboxSyncTool__loop is event_loop assert new_sync_tool._ToolboxSyncTool__thread is mock_thread + + +def test_toolbox_sync_tool_secure_params_properties( + toolbox_sync_tool: ToolboxSyncTool, + mock_async_tool: MagicMock, +): + """Tests _secure_params and _bound_secure_params properties.""" + mock_async_tool._secure_params = [ + ParameterSchema( + name="sec_param", + type="string", + description="desc", + required=True, + ) + ] + mock_async_tool._bound_secure_params = {"bound_sec": "val"} + + assert len(toolbox_sync_tool._secure_params) == 1 + assert toolbox_sync_tool._secure_params[0].name == "sec_param" + assert toolbox_sync_tool._bound_secure_params == {"bound_sec": "val"} + + +def test_toolbox_sync_tool_bind_secure_param( + toolbox_sync_tool: ToolboxSyncTool, + mock_async_tool: MagicMock, + event_loop: asyncio.AbstractEventLoop, + mock_thread: MagicMock, +): + """Tests bind_secure_param delegates to the wrapped async tool.""" + new_mock_async_tool = mock_async_tool.bind_secure_params.return_value + new_mock_async_tool.__name__ = "bound_async_tool" + + new_sync_tool = toolbox_sync_tool.bind_secure_param("api_key", "secret123") + + mock_async_tool.bind_secure_params.assert_called_once_with({"api_key": "secret123"}) + assert isinstance(new_sync_tool, ToolboxSyncTool) + assert new_sync_tool._ToolboxSyncTool__async_tool is new_mock_async_tool + + +def test_toolbox_sync_tool_bind_secure_params( + toolbox_sync_tool: ToolboxSyncTool, + mock_async_tool: MagicMock, + event_loop: asyncio.AbstractEventLoop, + mock_thread: MagicMock, +): + """Tests bind_secure_params delegates to the wrapped async tool.""" + new_mock_async_tool = mock_async_tool.bind_secure_params.return_value + new_mock_async_tool.__name__ = "bound_async_tool" + + params_to_bind = {"api_key": "secret", "db_pass": "pass"} + new_sync_tool = toolbox_sync_tool.bind_secure_params(params_to_bind) + + mock_async_tool.bind_secure_params.assert_called_once_with(params_to_bind) + assert isinstance(new_sync_tool, ToolboxSyncTool) + assert new_sync_tool._ToolboxSyncTool__async_tool is new_mock_async_tool diff --git a/packages/toolbox-core/tests/test_tool.py b/packages/toolbox-core/tests/test_tool.py index e675f464c..3cc544ded 100644 --- a/packages/toolbox-core/tests/test_tool.py +++ b/packages/toolbox-core/tests/test_tool.py @@ -929,3 +929,306 @@ async def test_telemetry_does_not_collide_with_param_named_telemetry_attributes( assert args[0] == "weird_tool" assert args[1] == {"telemetry_attributes": "user-data"} assert kwargs["telemetry_attributes"] is attrs + + +# --- Secure Parameters Tests --- + + +@pytest.fixture +def sample_secure_params() -> list[ParameterSchema]: + return [ + ParameterSchema( + name="api_key", + type="string", + required=True, + description="API secret key", + ), + ParameterSchema( + name="db_password", + type="string", + required=False, + description="Database password", + ), + ] + + +@pytest.mark.asyncio +async def test_tool_with_secure_parameters( + sample_tool_params: list[ParameterSchema], + sample_secure_params: list[ParameterSchema], + sample_tool_description: str, +): + transport = MockTransport(HTTPS_BASE_URL) + tool = ToolboxTool( + transport=transport, + 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={}, + secure_params=sample_secure_params, + ) + + assert len(tool._secure_params) == 2 + assert tool._secure_params[0].name == "api_key" + assert tool._bound_secure_params == {} + # Secure parameters must NOT appear in public signature + sig_param_names = list(tool.__signature__.parameters.keys()) + assert "api_key" not in sig_param_names + assert "db_password" not in sig_param_names + + +@pytest.mark.asyncio +async def test_bind_secure_param_success( + sample_tool_params: list[ParameterSchema], + sample_secure_params: list[ParameterSchema], + sample_tool_description: str, +): + transport = MockTransport(HTTPS_BASE_URL) + tool = ToolboxTool( + transport=transport, + 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={}, + secure_params=sample_secure_params, + ) + + bound_tool = tool.bind_secure_param("api_key", "secret-123") + + assert len(bound_tool._secure_params) == 1 + assert bound_tool._secure_params[0].name == "db_password" + assert bound_tool._bound_secure_params == {"api_key": "secret-123"} + # Immutability: original tool unchanged + assert len(tool._secure_params) == 2 + assert tool._bound_secure_params == {} + + # Invoke tool and verify secure_arguments passed to transport + transport.tool_invoke_mock.return_value = "Success" + await bound_tool(message="hello", count=5) + + transport.tool_invoke_mock.assert_awaited_once_with( + TEST_TOOL_NAME, + {"message": "hello", "count": 5}, + {}, + secure_arguments={"api_key": "secret-123"}, + ) + + +@pytest.mark.asyncio +async def test_bind_secure_params_with_callable( + sample_tool_params: list[ParameterSchema], + sample_secure_params: list[ParameterSchema], + sample_tool_description: str, +): + transport = MockTransport(HTTPS_BASE_URL) + tool = ToolboxTool( + transport=transport, + 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={}, + secure_params=sample_secure_params, + ) + + async def get_db_pass(): + return "async-secret-pass" + + bound_tool = tool.bind_secure_params( + {"api_key": lambda: "dynamic-api-key", "db_password": get_db_pass} + ) + + transport.tool_invoke_mock.return_value = "Success" + await bound_tool(message="hello", count=1) + + transport.tool_invoke_mock.assert_awaited_once_with( + TEST_TOOL_NAME, + {"message": "hello", "count": 1}, + {}, + secure_arguments={ + "api_key": "dynamic-api-key", + "db_password": "async-secret-pass", + }, + ) + + +@pytest.mark.asyncio +async def test_bind_secure_param_chaining( + sample_tool_params: list[ParameterSchema], + sample_secure_params: list[ParameterSchema], + sample_tool_description: str, +): + transport = MockTransport(HTTPS_BASE_URL) + tool = ToolboxTool( + transport=transport, + 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={}, + secure_params=sample_secure_params, + ) + + bound_tool = tool.bind_secure_param("api_key", "key1").bind_secure_param( + "db_password", "pass1" + ) + assert len(bound_tool._secure_params) == 0 + assert bound_tool._bound_secure_params == { + "api_key": "key1", + "db_password": "pass1", + } + + +@pytest.mark.asyncio +async def test_bind_secure_param_already_bound( + sample_tool_params: list[ParameterSchema], + sample_secure_params: list[ParameterSchema], + sample_tool_description: str, +): + transport = MockTransport(HTTPS_BASE_URL) + tool = ToolboxTool( + transport=transport, + 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={}, + secure_params=sample_secure_params, + ) + + bound_tool = tool.bind_secure_param("api_key", "secret1") + with pytest.raises( + ValueError, + match="cannot re-bind secure parameter: secure parameter 'api_key' is already bound", + ): + bound_tool.bind_secure_param("api_key", "secret2") + + +@pytest.mark.asyncio +async def test_bind_param_collision_with_secure_param( + sample_tool_params: list[ParameterSchema], + sample_secure_params: list[ParameterSchema], + sample_tool_description: str, +): + transport = MockTransport(HTTPS_BASE_URL) + tool = ToolboxTool( + transport=transport, + 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={}, + secure_params=sample_secure_params, + ) + + # Calling bind_param on a secure parameter should raise ValueError with guidance + with pytest.raises( + ValueError, + match="parameter 'api_key' is a secure parameter; use bind_secure_param/bind_secure_params instead", + ): + tool.bind_param("api_key", "val") + + +@pytest.mark.asyncio +async def test_bind_secure_param_collision_with_regular_param( + sample_tool_params: list[ParameterSchema], + sample_secure_params: list[ParameterSchema], + sample_tool_description: str, +): + transport = MockTransport(HTTPS_BASE_URL) + tool = ToolboxTool( + transport=transport, + 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={}, + secure_params=sample_secure_params, + ) + + # Calling bind_secure_param on a regular parameter should raise ValueError with guidance + with pytest.raises( + ValueError, + match="parameter 'message' is a regular parameter; use bind_param/bind_params instead", + ): + tool.bind_secure_param("message", "val") + + +@pytest.mark.asyncio +async def test_bind_secure_param_unknown( + sample_tool_params: list[ParameterSchema], + sample_secure_params: list[ParameterSchema], + sample_tool_description: str, +): + transport = MockTransport(HTTPS_BASE_URL) + tool = ToolboxTool( + transport=transport, + 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={}, + secure_params=sample_secure_params, + ) + + with pytest.raises( + ValueError, + match="unable to bind secure parameters: no secure parameter named nonexistent", + ): + tool.bind_secure_param("nonexistent", "val") + + +@pytest.mark.asyncio +async def test_missing_required_secure_parameter_fast_fail( + sample_tool_params: list[ParameterSchema], + sample_secure_params: list[ParameterSchema], + sample_tool_description: str, +): + transport = MockTransport(HTTPS_BASE_URL) + tool = ToolboxTool( + transport=transport, + 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={}, + secure_params=sample_secure_params, + ) + + # api_key is required and not bound -> invoke should fail fast before transport call + with pytest.raises( + ValueError, + match=r"Missing required secure parameter\(s\) \['api_key'\] for tool 'sample_tool'", + ): + await tool(message="hello", count=1) + + transport.tool_invoke_mock.assert_not_called() diff --git a/packages/toolbox-core/tests/test_utils.py b/packages/toolbox-core/tests/test_utils.py index ec6ebb22b..90a2c15b7 100644 --- a/packages/toolbox-core/tests/test_utils.py +++ b/packages/toolbox-core/tests/test_utils.py @@ -27,6 +27,7 @@ identify_auth_requirements, params_to_pydantic_model, resolve_value, + validate_unused_requirements, warn_if_http_and_headers, ) @@ -508,3 +509,72 @@ def test_warn_if_http_and_headers_https(): warnings.simplefilter("always") warn_if_http_and_headers(url, headers) assert len(w) == 0 + + +def test_validate_unused_requirements_all_used(): + """Test when all provided requirements are used.""" + # Should not raise + validate_unused_requirements( + provided_auth_keys={"auth1"}, + provided_bound_keys={"bound1"}, + used_auth_keys={"auth1"}, + used_bound_keys={"bound1"}, + name="test_tool", + is_toolset=False, + provided_secure_keys={"sec1"}, + used_secure_keys={"sec1"}, + ) + + +def test_validate_unused_requirements_unused_secure_params_single_tool(): + """Test unused secure parameter error formatting for a single tool.""" + with pytest.raises( + ValueError, + match=r"Validation failed for tool 'my_tool': unused secure parameters: sec1, sec2\.", + ): + validate_unused_requirements( + provided_auth_keys=set(), + provided_bound_keys=set(), + used_auth_keys=set(), + used_bound_keys=set(), + name="my_tool", + is_toolset=False, + provided_secure_keys={"sec1", "sec2"}, + used_secure_keys=set(), + ) + + +def test_validate_unused_requirements_unused_secure_params_toolset(): + """Test unused secure parameter error formatting for a toolset.""" + with pytest.raises( + ValueError, + match=r"Validation failed for toolset 'my_toolset': unused secure parameters could not be applied to any tool: sec1\.", + ): + validate_unused_requirements( + provided_auth_keys=set(), + provided_bound_keys=set(), + used_auth_keys=set(), + used_bound_keys=set(), + name="my_toolset", + is_toolset=True, + provided_secure_keys={"sec1"}, + used_secure_keys=set(), + ) + + +def test_validate_unused_requirements_combined_unused(): + """Test combined unused auth, bound, and secure parameters.""" + with pytest.raises( + ValueError, + match=r"Validation failed for tool 'my_tool': unused auth tokens: auth1; unused bound parameters: bound1; unused secure parameters: sec1\.", + ): + validate_unused_requirements( + provided_auth_keys={"auth1"}, + provided_bound_keys={"bound1"}, + used_auth_keys=set(), + used_bound_keys=set(), + name="my_tool", + is_toolset=False, + provided_secure_keys={"sec1"}, + used_secure_keys=set(), + )