30 lines
1.1 KiB
Python
30 lines
1.1 KiB
Python
import asyncio
|
|
import ipaddress
|
|
import socket
|
|
from urllib.parse import urlsplit
|
|
|
|
|
|
class UnsafeDestination(ValueError):
|
|
pass
|
|
|
|
|
|
async def validate_destination(url: str) -> None:
|
|
parsed = urlsplit(url)
|
|
if parsed.scheme not in {"http", "https"} or not parsed.hostname:
|
|
raise UnsafeDestination("only HTTP(S) URLs with a hostname are allowed")
|
|
if parsed.username or parsed.password:
|
|
raise UnsafeDestination("URL credentials are not allowed")
|
|
try:
|
|
literal = ipaddress.ip_address(parsed.hostname)
|
|
addresses = [literal]
|
|
except ValueError:
|
|
loop = asyncio.get_running_loop()
|
|
try:
|
|
infos = await loop.getaddrinfo(parsed.hostname, parsed.port,
|
|
type=socket.SOCK_STREAM)
|
|
except socket.gaierror as exc:
|
|
raise UnsafeDestination("destination DNS resolution failed") from exc
|
|
addresses = list({ipaddress.ip_address(info[4][0]) for info in infos})
|
|
if not addresses or any(not address.is_global for address in addresses):
|
|
raise UnsafeDestination("destination resolves to a non-public address")
|