import ipaddress import socket from typing import Any from urllib.parse import urlparse import requests def _is_url_safe(url: str) -> tuple[bool, str]: """Check if a URL is safe to request (not targeting private/internal networks).""" try: parsed = urlparse(url) hostname = parsed.hostname if not hostname: return False, "Could not parse hostname from URL" try: addr_infos = socket.getaddrinfo(hostname, None) except socket.gaierror: return False, f"Could not resolve hostname: {hostname}" for addr_info in addr_infos: ip_str = addr_info[4][0] try: ip = ipaddress.ip_address(ip_str) except ValueError: continue if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved: return False, f"URL resolves to blocked address: {ip_str}" return True, "" except Exception as e: # noqa: BLE001 return False, f"URL validation error: {e}" def _blocked_response(url: str, reason: str) -> dict[str, Any]: return { "success": False, "status_code": 0, "headers": {}, "content": f"Request blocked: {reason}", "url": url, } def http_request( url: str, method: str = "GET", headers: dict[str, str] | None = None, data: str | dict | None = None, params: dict[str, str] | None = None, timeout: int = 30, ) -> dict[str, Any]: """Make HTTP requests to APIs and web services. Args: url: Target URL method: HTTP method (GET, POST, PUT, DELETE, etc.) headers: HTTP headers to include data: Request body data (string or dict) params: URL query parameters timeout: Request timeout in seconds Returns: Dictionary with response data including status, headers, and content """ is_safe, reason = _is_url_safe(url) if not is_safe: return _blocked_response(url, reason) try: kwargs: dict[str, Any] = {} if headers: kwargs["headers"] = headers if params: kwargs["params"] = params if data: if isinstance(data, dict): kwargs["json"] = data else: kwargs["data"] = data response = requests.request(method.upper(), url, timeout=timeout, **kwargs) try: content = response.json() except (ValueError, requests.exceptions.JSONDecodeError): content = response.text return { "success": response.status_code < 400, "status_code": response.status_code, "headers": dict(response.headers), "content": content, "url": response.url, } except requests.exceptions.Timeout: return { "success": False, "status_code": 0, "headers": {}, "content": f"Request timed out after {timeout} seconds", "url": url, } except requests.exceptions.RequestException as e: return { "success": False, "status_code": 0, "headers": {}, "content": f"Request error: {e!s}", "url": url, }