62 lines
2.1 KiB
Python
62 lines
2.1 KiB
Python
import asyncio
|
|
import ipaddress
|
|
import socket
|
|
from collections.abc import Awaitable, Callable
|
|
from urllib.parse import urlsplit, urlunsplit
|
|
|
|
|
|
class UnsafeTargetError(ValueError):
|
|
pass
|
|
|
|
|
|
Resolver = Callable[[str, int], Awaitable[list[str]]]
|
|
|
|
|
|
async def resolve_addresses(host: str, port: int) -> list[str]:
|
|
loop = asyncio.get_running_loop()
|
|
records = await loop.run_in_executor(
|
|
None, lambda: socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)
|
|
)
|
|
return sorted({record[4][0] for record in records})
|
|
|
|
|
|
async def assert_safe_url(
|
|
url: str, allowed_ports: frozenset[int], resolver: Resolver = resolve_addresses
|
|
) -> str:
|
|
parsed = urlsplit(url)
|
|
if parsed.scheme not in {"http", "https"} or not parsed.hostname:
|
|
raise UnsafeTargetError("only HTTP(S) targets with a hostname are allowed")
|
|
if parsed.username is not None or parsed.password is not None:
|
|
raise UnsafeTargetError("target credentials are not allowed")
|
|
try:
|
|
port = parsed.port or (443 if parsed.scheme == "https" else 80)
|
|
except ValueError as exc:
|
|
raise UnsafeTargetError("target port is invalid") from exc
|
|
if port not in allowed_ports:
|
|
raise UnsafeTargetError("target port is not allowed")
|
|
try:
|
|
addresses = await resolver(parsed.hostname, port)
|
|
except (OSError, UnicodeError) as exc:
|
|
raise UnsafeTargetError("target hostname could not be resolved") from exc
|
|
if not addresses:
|
|
raise UnsafeTargetError("target hostname returned no addresses")
|
|
for value in addresses:
|
|
try:
|
|
address = ipaddress.ip_address(value)
|
|
except ValueError as exc:
|
|
raise UnsafeTargetError("resolver returned an invalid address") from exc
|
|
if not address.is_global:
|
|
raise UnsafeTargetError("target resolves to a non-public address")
|
|
return url
|
|
|
|
|
|
def redacted_url(url: str) -> str:
|
|
parsed = urlsplit(url)
|
|
host = parsed.hostname or "invalid"
|
|
try:
|
|
port = parsed.port
|
|
except ValueError:
|
|
port = None
|
|
netloc = f"{host}:{port}" if port else host
|
|
return urlunsplit((parsed.scheme, netloc, parsed.path, "", ""))
|