Harden protection with pinned requests and path-param encoding (#2486)

* Harden protection with pinned requests and path-param encoding

* fix: domain pinning

* fix: tests

* fix: html test 2

---------

Co-authored-by: Alex <a@tushynski.me>

tests/agents/tools/test_api_tool.py

@@ -81,104 +81,103 @@ class TestAPIToolInit:
 @pytest.mark.unit
 class TestMakeApiCall:
 
-    @patch("application.agents.tools.api_tool.validate_url")
-    @patch("application.agents.tools.api_tool.requests.get")
-    def test_successful_get(self, mock_get, mock_validate, get_tool):
+    @patch("application.agents.tools.api_tool.pinned_request")
+    def test_successful_get(self, mock_pinned, get_tool):
         mock_resp = MagicMock()
         mock_resp.status_code = 200
         mock_resp.headers = {"Content-Type": "application/json"}
         mock_resp.json.return_value = {"result": "ok"}
         mock_resp.content = b'{"result":"ok"}'
-        mock_get.return_value = mock_resp
+        mock_pinned.return_value = mock_resp
 
         result = get_tool.execute_action("any_action")
 
         assert result["status_code"] == 200
         assert result["data"] == {"result": "ok"}
         assert result["message"] == "API call successful."
+        assert mock_pinned.call_args[0][0] == "GET"
 
-    @patch("application.agents.tools.api_tool.validate_url")
-    @patch("application.agents.tools.api_tool.requests.post")
-    def test_successful_post(self, mock_post, mock_validate, post_tool):
+    @patch("application.agents.tools.api_tool.pinned_request")
+    def test_successful_post(self, mock_pinned, post_tool):
         mock_resp = MagicMock()
         mock_resp.status_code = 201
         mock_resp.headers = {"Content-Type": "application/json"}
         mock_resp.json.return_value = {"id": 1}
         mock_resp.content = b'{"id":1}'
-        mock_post.return_value = mock_resp
+        mock_pinned.return_value = mock_resp
 
         result = post_tool.execute_action("create", name="test")
         assert result["status_code"] == 201
+        assert mock_pinned.call_args[0][0] == "POST"
 
-    @patch("application.agents.tools.api_tool.validate_url")
-    @patch("application.agents.tools.api_tool.requests.put")
-    def test_put_method(self, mock_put, mock_validate):
+    @patch("application.agents.tools.api_tool.pinned_request")
+    def test_put_method(self, mock_pinned):
         tool = APITool(config={"url": "https://example.com/item/1", "method": "PUT"})
         mock_resp = MagicMock()
         mock_resp.status_code = 200
         mock_resp.headers = {"Content-Type": "application/json"}
         mock_resp.json.return_value = {}
         mock_resp.content = b'{}'
-        mock_put.return_value = mock_resp
+        mock_pinned.return_value = mock_resp
 
         result = tool.execute_action("update", name="new")
         assert result["status_code"] == 200
+        assert mock_pinned.call_args[0][0] == "PUT"
 
-    @patch("application.agents.tools.api_tool.validate_url")
-    @patch("application.agents.tools.api_tool.requests.delete")
-    def test_delete_method(self, mock_delete, mock_validate):
+    @patch("application.agents.tools.api_tool.pinned_request")
+    def test_delete_method(self, mock_pinned):
         tool = APITool(config={"url": "https://example.com/item/1", "method": "DELETE"})
         mock_resp = MagicMock()
         mock_resp.status_code = 204
         mock_resp.headers = {"Content-Type": "text/plain"}
         mock_resp.content = b''
-        mock_delete.return_value = mock_resp
+        mock_pinned.return_value = mock_resp
 
         result = tool.execute_action("delete")
         assert result["status_code"] == 204
+        assert mock_pinned.call_args[0][0] == "DELETE"
 
-    @patch("application.agents.tools.api_tool.validate_url")
-    @patch("application.agents.tools.api_tool.requests.patch")
-    def test_patch_method(self, mock_patch, mock_validate):
+    @patch("application.agents.tools.api_tool.pinned_request")
+    def test_patch_method(self, mock_pinned):
         tool = APITool(config={"url": "https://example.com/item/1", "method": "PATCH"})
         mock_resp = MagicMock()
         mock_resp.status_code = 200
         mock_resp.headers = {"Content-Type": "application/json"}
         mock_resp.json.return_value = {"patched": True}
         mock_resp.content = b'{"patched":true}'
-        mock_patch.return_value = mock_resp
+        mock_pinned.return_value = mock_resp
 
         result = tool.execute_action("patch", field="val")
         assert result["status_code"] == 200
+        assert mock_pinned.call_args[0][0] == "PATCH"
 
-    @patch("application.agents.tools.api_tool.validate_url")
-    @patch("application.agents.tools.api_tool.requests.head")
-    def test_head_method(self, mock_head, mock_validate):
+    @patch("application.agents.tools.api_tool.pinned_request")
+    def test_head_method(self, mock_pinned):
         tool = APITool(config={"url": "https://example.com", "method": "HEAD"})
         mock_resp = MagicMock()
         mock_resp.status_code = 200
         mock_resp.headers = {"Content-Type": "text/html"}
         mock_resp.content = b''
-        mock_head.return_value = mock_resp
+        mock_pinned.return_value = mock_resp
 
         result = tool.execute_action("check")
         assert result["status_code"] == 200
+        assert mock_pinned.call_args[0][0] == "HEAD"
 
-    @patch("application.agents.tools.api_tool.validate_url")
-    @patch("application.agents.tools.api_tool.requests.options")
-    def test_options_method(self, mock_options, mock_validate):
+    @patch("application.agents.tools.api_tool.pinned_request")
+    def test_options_method(self, mock_pinned):
         tool = APITool(config={"url": "https://example.com", "method": "OPTIONS"})
         mock_resp = MagicMock()
         mock_resp.status_code = 200
         mock_resp.headers = {"Content-Type": "text/plain"}
         mock_resp.content = b''
-        mock_options.return_value = mock_resp
+        mock_pinned.return_value = mock_resp
 
         result = tool.execute_action("options")
         assert result["status_code"] == 200
+        assert mock_pinned.call_args[0][0] == "OPTIONS"
 
-    @patch("application.agents.tools.api_tool.validate_url")
-    def test_unsupported_method(self, mock_validate):
+    def test_unsupported_method(self):
         tool = APITool(config={"url": "https://example.com", "method": "CUSTOM"})
         result = tool.execute_action("any")
         assert result["status_code"] is None
@@ -193,19 +192,18 @@ class TestMakeApiCall:
 @pytest.mark.unit
 class TestSSRFValidation:
 
-    @patch("application.agents.tools.api_tool.validate_url")
-    def test_ssrf_blocked_initial_url(self, mock_validate, get_tool):
-        from application.core.url_validation import SSRFError
+    @patch("application.agents.tools.api_tool.pinned_request")
+    def test_ssrf_blocked(self, mock_pinned, get_tool):
+        from application.security.safe_url import UnsafeUserUrlError
 
-        mock_validate.side_effect = SSRFError("blocked")
+        mock_pinned.side_effect = UnsafeUserUrlError("blocked")
         result = get_tool.execute_action("any")
         assert result["status_code"] is None
         assert "URL validation error" in result["message"]
 
-    @patch("application.agents.tools.api_tool.validate_url")
-    @patch("application.agents.tools.api_tool.requests.get")
-    def test_ssrf_blocked_after_param_substitution(self, mock_get, mock_validate):
-        from application.core.url_validation import SSRFError
+    @patch("application.agents.tools.api_tool.pinned_request")
+    def test_ssrf_blocked_with_path_params(self, mock_pinned):
+        from application.security.safe_url import UnsafeUserUrlError
 
         tool = APITool(config={
             "url": "https://api.example.com/{host}/data",
@@ -213,14 +211,7 @@ class TestSSRFValidation:
             "query_params": {"host": "169.254.169.254"},
         })
 
-        call_count = [0]
-
-        def side_effect(url):
-            call_count[0] += 1
-            if call_count[0] == 2:
-                raise SSRFError("blocked after substitution")
-
-        mock_validate.side_effect = side_effect
+        mock_pinned.side_effect = UnsafeUserUrlError("blocked")
         result = tool.execute_action("any")
         assert result["status_code"] is None
         assert "URL validation error" in result["message"]
@@ -234,40 +225,36 @@ class TestSSRFValidation:
 @pytest.mark.unit
 class TestErrorHandling:
 
-    @patch("application.agents.tools.api_tool.validate_url")
-    @patch("application.agents.tools.api_tool.requests.get")
-    def test_timeout_error(self, mock_get, mock_validate, get_tool):
-        mock_get.side_effect = requests.exceptions.Timeout()
+    @patch("application.agents.tools.api_tool.pinned_request")
+    def test_timeout_error(self, mock_pinned, get_tool):
+        mock_pinned.side_effect = requests.exceptions.Timeout()
         result = get_tool.execute_action("any")
         assert result["status_code"] is None
         assert "timeout" in result["message"].lower()
 
-    @patch("application.agents.tools.api_tool.validate_url")
-    @patch("application.agents.tools.api_tool.requests.get")
-    def test_connection_error(self, mock_get, mock_validate, get_tool):
-        mock_get.side_effect = requests.exceptions.ConnectionError("refused")
+    @patch("application.agents.tools.api_tool.pinned_request")
+    def test_connection_error(self, mock_pinned, get_tool):
+        mock_pinned.side_effect = requests.exceptions.ConnectionError("refused")
         result = get_tool.execute_action("any")
         assert result["status_code"] is None
         assert "Connection error" in result["message"]
 
-    @patch("application.agents.tools.api_tool.validate_url")
-    @patch("application.agents.tools.api_tool.requests.get")
-    def test_http_error_with_json(self, mock_get, mock_validate, get_tool):
+    @patch("application.agents.tools.api_tool.pinned_request")
+    def test_http_error_with_json(self, mock_pinned, get_tool):
         mock_resp = MagicMock()
         mock_resp.status_code = 422
         mock_resp.json.return_value = {"error": "invalid_field"}
         mock_resp.raise_for_status.side_effect = requests.exceptions.HTTPError(
             response=mock_resp
         )
-        mock_get.return_value = mock_resp
+        mock_pinned.return_value = mock_resp
 
         result = get_tool.execute_action("any")
         assert result["status_code"] == 422
         assert result["data"] == {"error": "invalid_field"}
 
-    @patch("application.agents.tools.api_tool.validate_url")
-    @patch("application.agents.tools.api_tool.requests.get")
-    def test_http_error_non_json_body(self, mock_get, mock_validate, get_tool):
+    @patch("application.agents.tools.api_tool.pinned_request")
+    def test_http_error_non_json_body(self, mock_pinned, get_tool):
         mock_resp = MagicMock()
         mock_resp.status_code = 404
         mock_resp.text = "Not Found"
@@ -275,29 +262,26 @@ class TestErrorHandling:
         mock_resp.raise_for_status.side_effect = requests.exceptions.HTTPError(
             response=mock_resp
         )
-        mock_get.return_value = mock_resp
+        mock_pinned.return_value = mock_resp
 
         result = get_tool.execute_action("any")
         assert result["status_code"] == 404
         assert result["data"] == "Not Found"
 
-    @patch("application.agents.tools.api_tool.validate_url")
-    @patch("application.agents.tools.api_tool.requests.get")
-    def test_request_exception(self, mock_get, mock_validate, get_tool):
-        mock_get.side_effect = requests.exceptions.RequestException("something")
+    @patch("application.agents.tools.api_tool.pinned_request")
+    def test_request_exception(self, mock_pinned, get_tool):
+        mock_pinned.side_effect = requests.exceptions.RequestException("something")
         result = get_tool.execute_action("any")
         assert "API call failed" in result["message"]
 
-    @patch("application.agents.tools.api_tool.validate_url")
-    @patch("application.agents.tools.api_tool.requests.get")
-    def test_unexpected_exception(self, mock_get, mock_validate, get_tool):
-        mock_get.side_effect = RuntimeError("unexpected")
+    @patch("application.agents.tools.api_tool.pinned_request")
+    def test_unexpected_exception(self, mock_pinned, get_tool):
+        mock_pinned.side_effect = RuntimeError("unexpected")
         result = get_tool.execute_action("any")
         assert "Unexpected error" in result["message"]
 
-    @patch("application.agents.tools.api_tool.validate_url")
-    @patch("application.agents.tools.api_tool.requests.post")
-    def test_body_serialization_error(self, mock_post, mock_validate):
+    @patch("application.agents.tools.api_tool.pinned_request")
+    def test_body_serialization_error(self, mock_pinned):
         tool = APITool(config={
             "url": "https://example.com",
             "method": "POST",
@@ -320,9 +304,8 @@ class TestErrorHandling:
 @pytest.mark.unit
 class TestPathParamSubstitution:
 
-    @patch("application.agents.tools.api_tool.validate_url")
-    @patch("application.agents.tools.api_tool.requests.get")
-    def test_path_params_substituted(self, mock_get, mock_validate):
+    @patch("application.agents.tools.api_tool.pinned_request")
+    def test_path_params_substituted(self, mock_pinned):
         tool = APITool(config={
             "url": "https://api.example.com/users/{user_id}/posts/{post_id}",
             "method": "GET",
@@ -333,17 +316,16 @@ class TestPathParamSubstitution:
         mock_resp.headers = {"Content-Type": "application/json"}
         mock_resp.json.return_value = []
         mock_resp.content = b'[]'
-        mock_get.return_value = mock_resp
+        mock_pinned.return_value = mock_resp
 
         tool.execute_action("get")
 
-        called_url = mock_get.call_args[0][0]
+        called_url = mock_pinned.call_args[0][1]
         assert "/users/42/posts/7" in called_url
         assert "{user_id}" not in called_url
 
-    @patch("application.agents.tools.api_tool.validate_url")
-    @patch("application.agents.tools.api_tool.requests.get")
-    def test_remaining_query_params_appended(self, mock_get, mock_validate):
+    @patch("application.agents.tools.api_tool.pinned_request")
+    def test_remaining_query_params_appended(self, mock_pinned):
         tool = APITool(config={
             "url": "https://api.example.com/items",
             "method": "GET",
@@ -354,19 +336,16 @@ class TestPathParamSubstitution:
         mock_resp.headers = {"Content-Type": "application/json"}
         mock_resp.json.return_value = []
         mock_resp.content = b'[]'
-        mock_get.return_value = mock_resp
+        mock_pinned.return_value = mock_resp
 
         tool.execute_action("get")
 
-        called_url = mock_get.call_args[0][0]
+        called_url = mock_pinned.call_args[0][1]
         assert "page=2" in called_url
         assert "limit=10" in called_url
 
-    @patch("application.agents.tools.api_tool.validate_url")
-    @patch("application.agents.tools.api_tool.requests.get")
-    def test_query_params_append_with_existing_query_string(
-        self, mock_get, mock_validate
-    ):
+    @patch("application.agents.tools.api_tool.pinned_request")
+    def test_query_params_append_with_existing_query_string(self, mock_pinned):
         tool = APITool(config={
             "url": "https://api.example.com/items?existing=true",
             "method": "GET",
@@ -377,27 +356,65 @@ class TestPathParamSubstitution:
         mock_resp.headers = {"Content-Type": "application/json"}
         mock_resp.json.return_value = []
         mock_resp.content = b'[]'
-        mock_get.return_value = mock_resp
+        mock_pinned.return_value = mock_resp
 
         tool.execute_action("get")
 
-        called_url = mock_get.call_args[0][0]
+        called_url = mock_pinned.call_args[0][1]
         assert "&page=1" in called_url
 
-    @patch("application.agents.tools.api_tool.validate_url")
-    @patch("application.agents.tools.api_tool.requests.post")
-    def test_empty_body_no_serialization(self, mock_post, mock_validate):
+    @patch("application.agents.tools.api_tool.pinned_request")
+    def test_empty_body_no_serialization(self, mock_pinned):
         tool = APITool(config={"url": "https://example.com", "method": "POST"})
         mock_resp = MagicMock()
         mock_resp.status_code = 200
         mock_resp.headers = {"Content-Type": "application/json"}
         mock_resp.json.return_value = {}
         mock_resp.content = b'{}'
-        mock_post.return_value = mock_resp
+        mock_pinned.return_value = mock_resp
 
         result = tool.execute_action("create")
         assert result["status_code"] == 200
 
+    @patch("application.agents.tools.api_tool.pinned_request")
+    def test_path_params_are_url_encoded(self, mock_pinned):
+        tool = APITool(config={
+            "url": "https://api.example.com/users/{user_id}/profile",
+            "method": "GET",
+            "query_params": {"user_id": "../../admin"},
+        })
+        mock_resp = MagicMock()
+        mock_resp.status_code = 200
+        mock_resp.headers = {"Content-Type": "application/json"}
+        mock_resp.json.return_value = {}
+        mock_resp.content = b'{}'
+        mock_pinned.return_value = mock_resp
+
+        tool.execute_action("get")
+
+        called_url = mock_pinned.call_args[0][1]
+        assert "../../admin" not in called_url
+        assert "%2F" in called_url or "%2f" in called_url
+
+    @patch("application.agents.tools.api_tool.pinned_request")
+    def test_path_params_query_injection_encoded(self, mock_pinned):
+        tool = APITool(config={
+            "url": "https://api.example.com/items/{item_id}",
+            "method": "GET",
+            "query_params": {"item_id": "x?admin=true"},
+        })
+        mock_resp = MagicMock()
+        mock_resp.status_code = 200
+        mock_resp.headers = {"Content-Type": "application/json"}
+        mock_resp.json.return_value = {}
+        mock_resp.content = b'{}'
+        mock_pinned.return_value = mock_resp
+
+        tool.execute_action("get")
+
+        called_url = mock_pinned.call_args[0][1]
+        assert "x?admin=true" not in called_url
+
 
 # =====================================================================
 # Parse Response
@@ -494,11 +511,8 @@ class TestAPIToolMetadata:
     def test_config_requirements_empty(self, get_tool):
         assert get_tool.get_config_requirements() == {}
 
-    @patch("application.agents.tools.api_tool.validate_url")
-    @patch("application.agents.tools.api_tool.requests.post")
-    def test_content_type_set_for_post_with_no_headers(
-        self, mock_post, mock_validate
-    ):
+    @patch("application.agents.tools.api_tool.pinned_request")
+    def test_content_type_set_for_post_with_no_headers(self, mock_pinned):
         tool = APITool(config={
             "url": "https://example.com",
             "method": "POST",
@@ -509,8 +523,8 @@ class TestAPIToolMetadata:
         mock_resp.headers = {"Content-Type": "application/json"}
         mock_resp.json.return_value = {}
         mock_resp.content = b'{}'
-        mock_post.return_value = mock_resp
+        mock_pinned.return_value = mock_resp
 
         tool.execute_action("create")
-        call_headers = mock_post.call_args[1]["headers"]
+        call_headers = mock_pinned.call_args.kwargs["headers"]
         assert "Content-Type" in call_headers