56 lines
1.9 KiB
Python
56 lines
1.9 KiB
Python
import asyncio
|
|
import ipaddress
|
|
import socket
|
|
from collections.abc import Awaitable, Callable
|
|
from urllib.parse import urlsplit, urlunsplit
|
|
|
|
Resolver = Callable[[str, int], Awaitable[list[str]]]
|
|
|
|
|
|
class UnsafeTarget(ValueError):
|
|
pass
|
|
|
|
|
|
def redact_url(url: str) -> str:
|
|
parts = urlsplit(url)
|
|
host = parts.hostname or "invalid"
|
|
if ":" in host:
|
|
host = f"[{host}]"
|
|
port = f":{parts.port}" if parts.port else ""
|
|
return urlunsplit((parts.scheme, f"{host}{port}", parts.path, "", ""))
|
|
|
|
|
|
async def system_resolver(host: str, port: int) -> list[str]:
|
|
def resolve() -> list[str]:
|
|
rows = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)
|
|
return list({row[4][0] for row in rows})
|
|
|
|
return await asyncio.to_thread(resolve)
|
|
|
|
|
|
async def validate_target(url: str, resolver: Resolver = system_resolver) -> None:
|
|
parts = urlsplit(url)
|
|
if parts.scheme not in {"http", "https"} or not parts.hostname:
|
|
raise UnsafeTarget("invalid_scheme_or_host")
|
|
if parts.username is not None or parts.password is not None:
|
|
raise UnsafeTarget("credentials_not_allowed")
|
|
host = parts.hostname.rstrip(".").lower()
|
|
if host == "localhost" or host.endswith(".localhost"):
|
|
raise UnsafeTarget("non_global_target")
|
|
port = parts.port or (443 if parts.scheme == "https" else 80)
|
|
try:
|
|
literal = ipaddress.ip_address(host)
|
|
addresses = [str(literal)]
|
|
except ValueError:
|
|
try:
|
|
addresses = await resolver(host, port)
|
|
except (OSError, socket.gaierror) as exc:
|
|
raise UnsafeTarget("dns_resolution_failed") from exc
|
|
if not addresses:
|
|
raise UnsafeTarget("dns_resolution_failed")
|
|
try:
|
|
if any(not ipaddress.ip_address(address).is_global for address in addresses):
|
|
raise UnsafeTarget("non_global_target")
|
|
except ValueError as exc:
|
|
raise UnsafeTarget("invalid_dns_answer") from exc
|