"""Universal HTTP request tool supporting any method, headers, cookies, and data."""
import httpx
import jsonutil as json
import logging
from typing import Optional
from tools._safe_http import (
assert_safe_http_url,
assert_safe_socks_proxy_url,
safe_http_headers,
safe_http_request,
safe_httpx_client,
)
from tools.alter_privileges import PRIVILEGES, has_privilege
logger = logging.getLogger(__name__)
VALID_METHODS = {"GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"}
TOOL_NAME = "http_request"
TOOL_DESCRIPTION = (
"Make an HTTP request with any method, headers, cookies, and body "
"to any URL. Returns status code, headers, body, and timing."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"url": {
"type": "string",
"description": "The URL to send the request to.",
},
"method": {
"type": "string",
"enum": ["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"],
"description": "HTTP method (default: GET).",
},
"headers": {
"type": "string",
"description": (
'JSON string of headers, e.g. \'{"Authorization": "Bearer token"}\'.'
),
},
"cookies": {
"type": "string",
"description": 'JSON string of cookies, e.g. \'{"session_id": "abc123"}\'.',
},
"data": {
"type": "string",
"description": "Raw body content as string.",
},
"json_data": {
"type": "string",
"description": (
"JSON string to send as request body. "
"Automatically sets Content-Type to application/json."
),
},
"form_data": {
"type": "string",
"description": (
"JSON string of form fields for "
"application/x-www-form-urlencoded encoding."
),
},
"params": {
"type": "string",
"description": 'JSON string of URL query parameters, e.g. \'{"page": "1"}\'.',
},
"timeout": {
"type": "number",
"description": "Request timeout in seconds (default: 30).",
},
"verify_ssl": {
"type": "boolean",
"description": "Verify SSL certificates (default: true).",
},
"follow_redirects": {
"type": "boolean",
"description": "Follow HTTP redirects (default: true).",
},
"proxy": {
"type": "string",
"description": (
"Optional SOCKS5 proxy URL, e.g. socks5h://127.0.0.1:9050 (Tor). "
"Omit for a direct connection."
),
},
},
"required": ["url"],
}
def _parse_json_param(name: str, value: Optional[str], expect_dict: bool = True):
"""Parse a JSON-string tool argument into a Python value.
Helper that decodes the string-typed JSON arguments (``headers``,
``cookies``, ``params``, ``form_data``) the model passes to this tool and,
when ``expect_dict`` is set, enforces that the result is an object. It folds
both the empty-input default and any decode/shape problem into a uniform
two-tuple so the caller can ``return err`` on failure without per-parameter
error handling.
Called only by ``run`` in this module; there are no external callers.
Args:
name: Parameter name, used purely for error messages.
value: The raw JSON string to parse, or a falsy value for "not provided".
expect_dict: When ``True``, require the parsed value to be a dict and
also use ``{}`` as the empty default; when ``False``, allow any JSON
type and use ``None`` as the empty default.
Returns:
A ``(parsed, error)`` tuple. On success ``error`` is ``None`` and
``parsed`` holds the decoded value (or the empty default). On failure
``parsed`` is ``None`` and ``error`` is a ready-to-return JSON error
string describing the invalid JSON or wrong type.
"""
if not value:
return ({} if expect_dict else None), None
try:
parsed = json.loads(value)
if expect_dict and not isinstance(parsed, dict):
return None, json.dumps(
{
"error": f"{name} must be a JSON object",
"details": f"Got {type(parsed).__name__} instead of dict",
}
)
return parsed, None
except json.JSONDecodeError as e:
return None, json.dumps(
{
"error": f"Invalid JSON in {name} parameter",
"details": str(e),
}
)
[docs]
async def run(
url: str,
method: str = "GET",
headers: Optional[str] = None,
cookies: Optional[str] = None,
data: Optional[str] = None,
json_data: Optional[str] = None,
form_data: Optional[str] = None,
params: Optional[str] = None,
timeout: float = 30.0,
verify_ssl: bool = True,
follow_redirects: bool = True,
proxy: Optional[str] = None,
ctx=None,
) -> str:
"""Make an arbitrary HTTP request and return the response details.
Entry point for the ``http_request`` tool, a general-purpose client
supporting any standard method, custom headers/cookies/query params, and a
raw, JSON, or form body. It is heavily SSRF-hardened: the target URL is
validated and normalized with ``assert_safe_http_url``, outbound headers are
sanitized via ``safe_http_headers``, an optional SOCKS5 proxy is checked with
``assert_safe_socks_proxy_url``, and the request itself runs through
``safe_httpx_client`` plus ``safe_http_request`` (from ``tools._safe_http``),
which re-validates redirect hops. Disabling TLS verification is privileged:
``verify_ssl=False`` is honored only when the caller holds
``UNSANDBOXED_EXEC`` (checked via ``tools.alter_privileges.has_privilege``
against the Redis-backed store), otherwise it is forced back on. At most one
body type may be supplied, and the timeout is clamped to 1-300 seconds.
Dispatched by the tool runner in ``tools/__init__.py``, which calls this
module's ``run`` (``tool_def.handler(**arguments, ctx=ctx)``) for the
registered ``http_request`` tool; there are no direct internal callers.
Args:
url (str): Target URL; must start with ``http://`` or ``https://``.
method (str): HTTP method; validated against ``VALID_METHODS`` and
upper-cased. Defaults to ``GET``.
headers (Optional[str]): JSON object string of request headers.
cookies (Optional[str]): JSON object string of cookies.
data (Optional[str]): Raw request body sent verbatim.
json_data (Optional[str]): JSON string sent as a JSON body (sets the
JSON content type).
form_data (Optional[str]): JSON object string sent as URL-encoded form
fields.
params (Optional[str]): JSON object string of URL query parameters.
timeout (float): Request timeout in seconds; clamped to 1-300, default 30.
verify_ssl (bool): Whether to verify TLS certificates; forced ``True``
unless the caller has ``UNSANDBOXED_EXEC``.
follow_redirects (bool): Whether redirects are followed (up to 5).
proxy (Optional[str]): Optional validated SOCKS5 proxy URL
(``socks5``/``socks5h``); omit for a direct connection.
ctx: Tool execution context; supplies ``redis``, ``config``, and
``user_id`` for the TLS-off privilege check.
Returns:
str: JSON. On success an object with ``status_code``, ``headers``,
``body``, the final ``url``, and ``elapsed_ms``; on failure an
``{"error": ..., "details": ...}`` object (bad input, blocked URL,
timeout, connection failure, too many redirects, etc.).
"""
method = method.upper().strip()
if method not in VALID_METHODS:
return json.dumps(
{
"error": f"Invalid HTTP method: {method}",
"details": f"Valid methods are: {', '.join(sorted(VALID_METHODS))}",
}
)
if not url or not url.strip():
return json.dumps({"error": "URL is required and cannot be empty"})
url = url.strip()
if not url.startswith(("http://", "https://")):
return json.dumps(
{
"error": "Invalid URL format",
"details": "URL must start with http:// or https://",
}
)
try:
url = assert_safe_http_url(url)
except ValueError as exc:
return json.dumps({"error": "URL not allowed", "details": str(exc)})
if not verify_ssl:
allow_tls_off = False
if ctx is not None:
redis = getattr(ctx, "redis", None)
cfg = getattr(ctx, "config", None)
uid = getattr(ctx, "user_id", None)
if redis is not None and cfg is not None and uid is not None:
allow_tls_off = await has_privilege(
redis,
uid,
PRIVILEGES["UNSANDBOXED_EXEC"],
cfg,
)
if not allow_tls_off:
verify_ssl = True
parsed_headers, err = _parse_json_param("headers", headers)
if err:
return err
parsed_cookies, err = _parse_json_param("cookies", cookies)
if err:
return err
parsed_params, err = _parse_json_param("params", params)
if err:
return err
parsed_json_data = None
if json_data:
try:
parsed_json_data = json.loads(json_data)
except json.JSONDecodeError as e:
return json.dumps(
{
"error": "Invalid JSON in json_data parameter",
"details": str(e),
}
)
parsed_form_data, err = _parse_json_param("form_data", form_data)
if err:
return err
if not form_data:
parsed_form_data = None
try:
timeout = float(timeout)
if timeout <= 0:
timeout = 30.0
elif timeout > 300:
timeout = 300.0
except (TypeError, ValueError):
timeout = 30.0
request_kwargs = {
"method": method,
"url": url,
"timeout": timeout,
}
if parsed_headers:
request_kwargs["headers"] = safe_http_headers(parsed_headers)
if parsed_cookies:
request_kwargs["cookies"] = parsed_cookies
if parsed_params:
request_kwargs["params"] = parsed_params
body_count = sum(
[
data is not None,
parsed_json_data is not None,
parsed_form_data is not None,
]
)
if body_count > 1:
return json.dumps(
{
"error": "Multiple body types specified",
"details": "Use only one of: data, json_data, or form_data",
}
)
if parsed_json_data is not None:
request_kwargs["json"] = parsed_json_data
elif parsed_form_data is not None:
request_kwargs["data"] = parsed_form_data
elif data is not None:
request_kwargs["content"] = data
client_kwargs = {"verify": verify_ssl}
if proxy and str(proxy).strip():
try:
client_kwargs["proxy"] = assert_safe_socks_proxy_url(proxy.strip())
except ValueError as exc:
return json.dumps({"error": "Invalid proxy URL", "details": str(exc)})
try:
logger.info(f"HTTP {method} request to {url}")
req_kw = dict(request_kwargs)
req_kw.pop("url", None)
req_kw.pop("method", None)
max_r = 5 if follow_redirects else 0
async with safe_httpx_client(**client_kwargs) as client:
response = await safe_http_request(
client,
method,
url,
max_redirects=max_r,
**req_kw,
)
response_headers = dict(response.headers)
elapsed_ms = response.elapsed.total_seconds() * 1000
try:
body = response.text
except Exception:
body = "<binary content>"
result = {
"status_code": response.status_code,
"headers": response_headers,
"body": body,
"url": str(response.url),
"elapsed_ms": round(elapsed_ms, 2),
}
logger.info(
f"HTTP {method} {url} completed: {response.status_code} in {elapsed_ms:.2f}ms"
)
return json.dumps(result, ensure_ascii=False)
except httpx.TimeoutException:
return json.dumps(
{
"error": "Request timed out",
"details": f"No response received within {timeout} seconds",
}
)
except httpx.ConnectError as e:
return json.dumps({"error": "Connection failed", "details": str(e)})
except httpx.TooManyRedirects:
return json.dumps(
{
"error": "Too many redirects",
"details": "The request exceeded the maximum number of redirects",
}
)
except httpx.RequestError as e:
return json.dumps(
{
"error": f"Request error: {type(e).__name__}",
"details": str(e),
}
)
except Exception as e:
logger.error(f"Unexpected error in http_request: {e}", exc_info=True)
return json.dumps(
{
"error": f"Unexpected error: {type(e).__name__}",
"details": str(e),
}
)