39 lines
1.3 KiB
Python
39 lines
1.3 KiB
Python
import asyncio
|
|
import ipaddress
|
|
import socket
|
|
from collections.abc import Awaitable, Callable
|
|
from urllib.parse import urlsplit
|
|
|
|
Resolver = Callable[[str, int], Awaitable[list[str]]]
|
|
|
|
|
|
class UnsafeTarget(ValueError):
|
|
pass
|
|
|
|
|
|
async def system_resolver(host: str, port: int) -> list[str]:
|
|
loop = asyncio.get_running_loop()
|
|
records = await loop.getaddrinfo(host, port, type=socket.SOCK_STREAM)
|
|
return sorted({record[4][0] for record in records})
|
|
|
|
|
|
async def validate_target(url: str, resolver: Resolver = system_resolver) -> None:
|
|
parsed = urlsplit(url)
|
|
if parsed.scheme not in {"http", "https"} or not parsed.hostname:
|
|
raise UnsafeTarget("unsupported_target")
|
|
if parsed.username or parsed.password:
|
|
raise UnsafeTarget("credentials_not_allowed")
|
|
port = parsed.port or (443 if parsed.scheme == "https" else 80)
|
|
try:
|
|
addresses = await resolver(parsed.hostname, port)
|
|
except (OSError, socket.gaierror) as exc:
|
|
raise UnsafeTarget("dns_resolution_failed") from exc
|
|
if not addresses:
|
|
raise UnsafeTarget("dns_resolution_failed")
|
|
try:
|
|
safe = all(ipaddress.ip_address(address).is_global for address in addresses)
|
|
except ValueError as exc:
|
|
raise UnsafeTarget("dns_resolution_failed") from exc
|
|
if not safe:
|
|
raise UnsafeTarget("non_public_target")
|