src/http_executor.py

# SPDX-FileCopyrightText: Copyright (c) 2026 Abhinav Kalvacherla
# SPDX-License-Identifier: Apache-2.0
"""
Generic HTTP executor for client-supplied function/API definitions.

A function definition (supplied per-session, see session_config.py) is a dict:

    {
      "name": "get_weather",                       # required — tool name
      "description": "Current weather for a lat/lon.",
      "endpoint": "https://api.open-meteo.com/v1/forecast",  # required — may contain {placeholders}
      "method": "GET",                             # GET|POST|PUT|PATCH|DELETE (default GET)
      "param_location": "query",                   # query|body|path (default: query for GET, body otherwise)
      "param_locations": {"id": "path"},           # optional per-parameter override
      "headers": {"X-Api-Key": "..."},             # optional static request headers
      "static_params": {"current_weather": true},  # optional server-merged constants
      "parameters": { ...JSON schema... }          # OpenAI tool parameters schema
    }

`build_openai_tools` turns the {name, description, parameters} parts into the
OpenAI `tools` array given to the LLM. `execute_function` takes the LLM-extracted
args, builds the HTTP request per the spec, and returns the response text.

Security: the server makes requests to client-supplied URLs (an SSRF surface).
`_enforce_allowlist` restricts hosts and blocks private/loopback/link-local IPs.
"""

from __future__ import annotations

import ipaddress
import logging
import os
import socket
from urllib.parse import urlparse

import httpx

logger = logging.getLogger(__name__)

MAX_RESPONSE_CHARS = 1500
HTTP_TIMEOUT = 4.0   # below rag_timeout; the retriever's wait_for is the hard ceiling


def build_openai_tools(functions: list[dict]) -> list[dict]:
    """Map declarative function defs → OpenAI `tools` array for the LLM."""
    tools = []
    for f in functions:
        tools.append({
            "type": "function",
            "function": {
                "name": f["name"],
                "description": f.get("description", ""),
                "parameters": f.get("parameters") or {"type": "object", "properties": {}},
            },
        })
    return tools


def _default_location(spec: dict) -> str:
    return "query" if str(spec.get("method", "GET")).upper() == "GET" else "body"


def _split_args(spec: dict, args: dict) -> tuple[str, dict, dict | None]:
    """Route merged args into URL path / query / body per the spec."""
    loc = spec.get("param_location") or _default_location(spec)
    per = spec.get("param_locations") or {}
    merged = {**(spec.get("static_params") or {}), **(args or {})}

    url = spec["endpoint"]
    query: dict = {}
    body: dict = {}
    for k, v in merged.items():
        placeholder = "{%s}" % k
        where = per.get(k, "path" if placeholder in url else loc)
        if where == "path":
            url = url.replace(placeholder, str(v))
        elif where == "body":
            body[k] = v
        else:
            query[k] = v
    return url, query, (body or None)


def _enforce_allowlist(url: str, cfg) -> None:
    """Reject disallowed hosts and any host resolving to a non-public IP (SSRF guard)."""
    parsed = urlparse(url)
    if parsed.scheme not in ("http", "https"):
        raise PermissionError(f"scheme {parsed.scheme!r} not allowed")
    host = parsed.hostname or ""

    server_allow = {h for h in (os.environ.get("MOSHI_TOOL_ALLOWED_HOSTS") or "").split(",") if h}
    sess_allow = set(getattr(cfg, "allowed_hosts", None) or [])
    if server_allow and sess_allow:
        allow = server_allow & sess_allow
    else:
        allow = server_allow or sess_allow
    if allow and host not in allow:
        raise PermissionError(f"host {host!r} not in allowlist")

    # Block private / loopback / link-local / reserved targets regardless of allowlist.
    try:
        infos = socket.getaddrinfo(host, None)
    except socket.gaierror as exc:
        raise PermissionError(f"cannot resolve host {host!r}: {exc}") from exc
    for info in infos:
        ip = ipaddress.ip_address(info[4][0])
        if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved or ip.is_multicast:
            raise PermissionError(f"host {host!r} resolves to non-public IP {ip}")


async def execute_function(spec: dict, args: dict, cfg) -> str:
    """Build and make the HTTP call described by `spec` with `args`; return response text."""
    _enforce_allowlist(spec["endpoint"], cfg)
    method = str(spec.get("method", "GET")).upper()
    url, query, body = _split_args(spec, args)
    headers = spec.get("headers") or {}

    logger.info("[http] %s %s params=%s body=%s", method, url, query, body)
    async with httpx.AsyncClient(timeout=HTTP_TIMEOUT, follow_redirects=False) as client:
        resp = await client.request(method, url, params=query or None, json=body, headers=headers)
        resp.raise_for_status()
        text = resp.text
    return text[:MAX_RESPONSE_CHARS]