|
# retoor <retoor@molodetz.nl>
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import ipaddress
|
|
import socket
|
|
from typing import Any
|
|
from urllib.parse import urlparse
|
|
|
|
import httpx
|
|
|
|
NAT64_PREFIXES = (
|
|
ipaddress.ip_network("64:ff9b::/96"),
|
|
ipaddress.ip_network("64:ff9b:1::/48"),
|
|
)
|
|
|
|
|
|
class BlockedAddressError(ValueError):
|
|
pass
|
|
|
|
|
|
def effective_address(address: Any) -> Any:
|
|
if isinstance(address, ipaddress.IPv6Address):
|
|
if address.ipv4_mapped is not None:
|
|
return address.ipv4_mapped
|
|
for prefix in NAT64_PREFIXES:
|
|
if address in prefix:
|
|
return ipaddress.IPv4Address(int(address) & 0xFFFFFFFF)
|
|
return address
|
|
|
|
|
|
def is_blocked_address(address: Any) -> bool:
|
|
resolved = effective_address(address)
|
|
return (
|
|
resolved.is_private
|
|
or resolved.is_loopback
|
|
or resolved.is_link_local
|
|
or resolved.is_reserved
|
|
or resolved.is_multicast
|
|
or resolved.is_unspecified
|
|
)
|
|
|
|
|
|
async def guard_public_url(url: str, *, allow_private: bool = False) -> str:
|
|
parsed = urlparse(url)
|
|
if parsed.scheme not in ("http", "https"):
|
|
raise BlockedAddressError("Only http and https URLs are allowed.")
|
|
host = parsed.hostname
|
|
if not host:
|
|
raise BlockedAddressError("URL has no host.")
|
|
if allow_private:
|
|
return host
|
|
try:
|
|
infos = await asyncio.to_thread(socket.getaddrinfo, host, None)
|
|
except socket.gaierror as exc:
|
|
raise BlockedAddressError(f"Could not resolve host: {host}") from exc
|
|
for info in infos:
|
|
address = ipaddress.ip_address(info[4][0])
|
|
if is_blocked_address(address):
|
|
raise BlockedAddressError(
|
|
f"Refusing to reach a private or local address ({effective_address(address)})."
|
|
)
|
|
return host
|
|
|
|
|
|
class _GuardedTransport(httpx.AsyncHTTPTransport):
|
|
async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
|
|
await guard_public_url(str(request.url))
|
|
return await super().handle_async_request(request)
|
|
|
|
|
|
def guarded_async_client(**kwargs: Any) -> httpx.AsyncClient:
|
|
transport = _GuardedTransport()
|
|
return httpx.AsyncClient(transport=transport, **kwargs)
|