116 lines
3.2 KiB
Python
116 lines
3.2 KiB
Python
|
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,
|
||
|
|
}
|