feat: Introduce identifying used authz token getters

This PR adds the feature in `identify_required_authn_params` helper to determine which of the provided auth token getters are actively used to satisfy a tool's authorization requirements (as defined by the `authRequired` key in its manifest).

This is a foundational step towards future validation in `ToolboxTool.add_auth_token_getters`, which will ensure no configured auth token getters remain unused, thereby preventing potential misconfigurations.

Author: anubhav756

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

@@ -79,8 +79,11 @@ def __parse_tool(
             else:  # regular parameter
                 params.append(p)
 
-        authn_params, used_auth_keys = identify_required_authn_params(
-            authn_params, auth_token_getters.keys()
+        authn_params, _, used_auth_keys = identify_required_authn_params(
+            # TODO: Add schema.authRequired as second arg
+            authn_params,
+            [],
+            auth_token_getters.keys(),
         )
 
         tool = ToolboxTool(

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

@@ -299,7 +299,10 @@ def add_auth_token_getters(
         # create a read-only updated for params that are still required
         new_req_authn_params = MappingProxyType(
             identify_required_authn_params(
-                self.__required_authn_params, auth_token_getters.keys()
+                # TODO: Add authRequired
+                self.__required_authn_params,
+                [],
+                auth_token_getters.keys(),
             )[0]
         )
 

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

@@ -45,18 +45,23 @@ def create_func_docstring(description: str, params: Sequence[ParameterSchema]) -
 
 
 def identify_required_authn_params(
-    req_authn_params: Mapping[str, list[str]], auth_service_names: Iterable[str]
-) -> tuple[dict[str, list[str]], set[str]]:
+    req_authn_params: Mapping[str, list[str]],
+    req_authz_tokens: list[str],
+    auth_service_names: Iterable[str],
+) -> tuple[dict[str, list[str]], list[str], set[str]]:
     """
-    Identifies authentication parameters that are still required; because they
-        are not covered by the provided `auth_service_names`, and also returns a
-        set of all authentication services that were found to be matching.
+    Identifies authentication parameters and authorization tokens that are still
+    required because they are not covered by the provided `auth_service_names`.
+    Also returns a set of all authentication/authorization services from
+    `auth_service_names` that were found to be matching.
 
-        Args:
-            req_authn_params: A mapping of parameter names to lists of required
-                authentication services.
-            auth_service_names: An iterable of authentication service names for which
-                token getters are available.
+    Args:
+        req_authn_params: A mapping of parameter names to lists of required
+            authentication services for those parameters.
+        req_authz_tokens: A list of strings representing all authorization
+            tokens that are required to invoke the current tool.
+        auth_service_names: An iterable of authentication/authorization service
+            names for which token getters are available.
 
     Returns:
         A tuple containing:
@@ -70,20 +75,34 @@ def identify_required_authn_params(
               that were found to satisfy at least one authentication parameter's
               requirements or matched one of the `req_authz_tokens`.
     """
-    required_params: dict[str, list[str]] = {}
+    required_authn_params: dict[str, list[str]] = {}
     used_services: set[str] = set()
 
+    # find which of the required authn params are covered by available services.
     for param, services in req_authn_params.items():
+
         # if we don't have a token_getter for any of the services required by the param,
         # the param is still required
-        matched_services = [s for s in services if s in auth_service_names]
+        matched_authn_services = [s for s in services if s in auth_service_names]
 
-        if matched_services:
-            used_services.update(matched_services)
+        if matched_authn_services:
+            used_services.update(matched_authn_services)
         else:
-            required_params[param] = services
+            required_authn_params[param] = services
+
+    # find which of the required authz tokens are covered by available services.
+    matched_authz_services = [s for s in auth_service_names if s in req_authz_tokens]
+    required_authz_tokens: list[str] = []
+
+    # If a match is found, authorization is met (no remaining required tokens).
+    # Otherwise, all `req_authz_tokens` are still required. (Handles empty
+    # `req_authz_tokens` correctly, resulting in no required tokens).
+    if matched_authz_services:
+        used_services.update(matched_authz_services)
+    else:
+        required_authz_tokens = req_authz_tokens
 
-    return required_params, used_services
+    return required_authn_params, required_authz_tokens, used_services
 
 
 def params_to_pydantic_model(

packages/toolbox-core/tests/test_utils.py

@@ -83,10 +83,12 @@ def test_create_func_docstring_empty_description():
 
 
 def test_identify_required_authn_params_none_required():
-    """Test when no authentication parameters are required initially."""
-    req_authn_params = {}
+    """Test when no authentication parameters or authorization tokens are required initially."""
+    req_authn_params: dict[str, list[str]] = {}
+    req_authz_tokens: list[str] = []
     auth_service_names = ["service_a", "service_b"]
-    expected = {}
+    expected_params = {}
+    expected_authz: list[str] = []
     expected_used = set()
     result = identify_required_authn_params(
         req_authn_params, req_authz_tokens, auth_service_names
@@ -99,11 +101,12 @@ def test_identify_required_authn_params_none_required():
 
 
 def test_identify_required_authn_params_all_covered():
-    """Test when all required parameters are covered by available services."""
+    """Test when all required authn parameters are covered, no authz tokens."""
     req_authn_params = {
         "token_a": ["service_a"],
         "token_b": ["service_b", "service_c"],
     }
+    req_authz_tokens: list[str] = []
     auth_service_names = ["service_a", "service_b"]
     expected_params = {}
     expected_authz: list[str] = []
@@ -119,15 +122,16 @@ def test_identify_required_authn_params_all_covered():
 
 
 def test_identify_required_authn_params_some_covered():
-    """Test when some parameters are covered, and some are not."""
+    """Test when some authn parameters are covered, and some are not, no authz tokens."""
     req_authn_params = {
         "token_a": ["service_a"],
         "token_b": ["service_b", "service_c"],
         "token_d": ["service_d"],
         "token_e": ["service_e", "service_f"],
     }
+    req_authz_tokens: list[str] = []
     auth_service_names = ["service_a", "service_b"]
-    expected = {
+    expected_params = {
         "token_d": ["service_d"],
         "token_e": ["service_e", "service_f"],
     }
@@ -145,16 +149,18 @@ def test_identify_required_authn_params_some_covered():
 
 
 def test_identify_required_authn_params_none_covered():
-    """Test when none of the required parameters are covered."""
+    """Test when none of the required authn parameters are covered, no authz tokens."""
     req_authn_params = {
         "token_d": ["service_d"],
         "token_e": ["service_e", "service_f"],
     }
+    req_authz_tokens: list[str] = []
     auth_service_names = ["service_a", "service_b"]
-    expected = {
+    expected_params = {
         "token_d": ["service_d"],
         "token_e": ["service_e", "service_f"],
     }
+    expected_authz: list[str] = []
     expected_used = set()
     result = identify_required_authn_params(
         req_authn_params, req_authz_tokens, auth_service_names
@@ -167,16 +173,18 @@ def test_identify_required_authn_params_none_covered():
 
 
 def test_identify_required_authn_params_no_available_services():
-    """Test when no authentication services are available."""
+    """Test when no authn services are available, no authz tokens."""
     req_authn_params = {
         "token_a": ["service_a"],
         "token_b": ["service_b", "service_c"],
     }
-    auth_service_names = []
-    expected = {
+    req_authz_tokens: list[str] = []
+    auth_service_names: list[str] = []
+    expected_params = {
         "token_a": ["service_a"],
         "token_b": ["service_b", "service_c"],
     }
+    expected_authz: list[str] = []
     expected_used = set()
     result = identify_required_authn_params(
         req_authn_params, req_authz_tokens, auth_service_names
@@ -189,14 +197,16 @@ def test_identify_required_authn_params_no_available_services():
 
 
 def test_identify_required_authn_params_empty_services_for_param():
-    """Test edge case where a param requires an empty list of services."""
+    """Test edge case: authn param requires an empty list of services, no authz tokens."""
     req_authn_params = {
         "token_x": [],
     }
+    req_authz_tokens: list[str] = []
     auth_service_names = ["service_a"]
-    expected = {
+    expected_params = {
         "token_x": [],
     }
+    expected_authz: list[str] = []
     expected_used = set()
     result = identify_required_authn_params(
         req_authn_params, req_authz_tokens, auth_service_names