@@ -0,0 +1,70 @@
|
||||
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)
|
||||
Reference in New Issue
Block a user