71 lines
2.4 KiB
Python
71 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
import ipaddress
|
|
import socket
|
|
from collections.abc import Callable
|
|
from dataclasses import dataclass
|
|
from urllib.parse import urlsplit
|
|
|
|
Resolver = Callable[..., list[tuple]]
|
|
|
|
|
|
class UnsafeUrlError(ValueError):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ValidatedUrl:
|
|
url: str
|
|
hostname: str
|
|
port: int
|
|
addresses: tuple[str, ...]
|
|
|
|
|
|
def _extract_addresses(results: list[tuple]) -> tuple[str, ...]:
|
|
addresses: list[str] = []
|
|
for result in results:
|
|
sockaddr = result[4]
|
|
if sockaddr:
|
|
addresses.append(str(sockaddr[0]))
|
|
return tuple(dict.fromkeys(addresses))
|
|
|
|
|
|
def validate_public_url(
|
|
url: str,
|
|
*,
|
|
resolver: Resolver = socket.getaddrinfo,
|
|
allow_nonstandard_ports: bool = False,
|
|
) -> ValidatedUrl:
|
|
parts = urlsplit(url)
|
|
if parts.scheme.lower() not in {"http", "https"}:
|
|
raise UnsafeUrlError("Alleen http en https zijn toegestaan.")
|
|
if parts.username or parts.password:
|
|
raise UnsafeUrlError("URLs met ingebedde credentials zijn niet toegestaan.")
|
|
hostname = (parts.hostname or "").lower().rstrip(".")
|
|
if not hostname:
|
|
raise UnsafeUrlError("URL bevat geen hostname.")
|
|
if hostname == "localhost" or hostname.endswith(".localhost"):
|
|
raise UnsafeUrlError("Lokale hostnames zijn niet toegestaan.")
|
|
port = parts.port or (443 if parts.scheme.lower() == "https" else 80)
|
|
if not allow_nonstandard_ports and port not in {80, 443}:
|
|
raise UnsafeUrlError("Niet-standaard poorten zijn niet toegestaan.")
|
|
try:
|
|
direct_ip = ipaddress.ip_address(hostname.strip("[]"))
|
|
addresses = (str(direct_ip),)
|
|
except ValueError:
|
|
try:
|
|
results = resolver(hostname, port, type=socket.SOCK_STREAM)
|
|
except OSError as exc:
|
|
raise UnsafeUrlError("Hostname kan niet veilig worden opgelost.") from exc
|
|
addresses = _extract_addresses(results)
|
|
if not addresses:
|
|
raise UnsafeUrlError("Hostname leverde geen IP-adressen op.")
|
|
for raw in addresses:
|
|
try:
|
|
ip = ipaddress.ip_address(raw)
|
|
except ValueError as exc:
|
|
raise UnsafeUrlError("Ongeldig IP-adres na DNS-resolutie.") from exc
|
|
if not ip.is_global:
|
|
raise UnsafeUrlError(f"Niet-publiek IP-adres geblokkeerd: {ip}")
|
|
return ValidatedUrl(url=url, hostname=hostname, port=port, addresses=addresses)
|