77 lines
3.4 KiB
Python
77 lines
3.4 KiB
Python
"""Resolve once, reject non-public destinations, and keep the approved address."""
|
|
|
|
from dataclasses import dataclass
|
|
import ipaddress
|
|
import socket
|
|
from urllib.parse import urlsplit, urlunsplit
|
|
|
|
from .callback_policy import ALLOWED_CALLBACK_HOSTS
|
|
|
|
|
|
class CallbackURLValidationError(ValueError):
|
|
"""A callback URL failed validation."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CallbackTarget:
|
|
url: str
|
|
hostname: str
|
|
address: str
|
|
port: int
|
|
scheme: str
|
|
authority: str
|
|
|
|
|
|
def _public_address(value: str) -> bool:
|
|
address = ipaddress.ip_address(value)
|
|
if isinstance(address, ipaddress.IPv6Address):
|
|
if address.ipv4_mapped is not None:
|
|
return _public_address(str(address.ipv4_mapped))
|
|
if address.sixtofour or address.teredo or address in ipaddress.ip_network("64:ff9b::/96"):
|
|
return False
|
|
return address.is_global and not (address.is_multicast or address.is_reserved)
|
|
|
|
|
|
def validate_callback_url(callback_url: str) -> CallbackTarget:
|
|
value = (callback_url or "").strip()
|
|
if not value:
|
|
raise CallbackURLValidationError("callback_url is required")
|
|
if any(char.isspace() or ord(char) < 32 or ord(char) == 127 for char in value) or "\\" in value:
|
|
raise CallbackURLValidationError("callback_url contains invalid characters")
|
|
try:
|
|
parsed = urlsplit(value)
|
|
if parsed.scheme != "https":
|
|
raise CallbackURLValidationError("callback_url must use HTTPS")
|
|
if not parsed.hostname:
|
|
raise CallbackURLValidationError("callback_url must include a hostname")
|
|
if parsed.username is not None or parsed.password is not None:
|
|
raise CallbackURLValidationError("callback_url must not contain a username or password")
|
|
if parsed.fragment:
|
|
raise CallbackURLValidationError("callback_url must not contain a fragment")
|
|
hostname = parsed.hostname
|
|
if hostname.endswith("."):
|
|
hostname = hostname[:-1]
|
|
hostname = hostname.encode("idna").decode("ascii").lower()
|
|
# Check before DNS and HTTP; imported workflows cannot extend this policy.
|
|
if hostname not in ALLOWED_CALLBACK_HOSTS:
|
|
raise CallbackURLValidationError("callback_url host is not in the approved allowlist")
|
|
port = parsed.port if parsed.port is not None else 443
|
|
if port == 0:
|
|
raise CallbackURLValidationError("callback_url has an invalid port")
|
|
try:
|
|
addresses = [str(ipaddress.ip_address(hostname))]
|
|
except ValueError:
|
|
addresses = list(dict.fromkeys(item[4][0] for item in socket.getaddrinfo(
|
|
hostname, port, type=socket.SOCK_STREAM
|
|
)))
|
|
if not addresses or not all(_public_address(address) for address in addresses):
|
|
raise CallbackURLValidationError("callback_url must resolve only to public addresses")
|
|
host = f"[{hostname}]" if ":" in hostname else hostname
|
|
authority = f"{host}:{port}" if parsed.port is not None else host
|
|
normalized = urlunsplit((parsed.scheme, authority, parsed.path or "/", parsed.query, ""))
|
|
return CallbackTarget(normalized, hostname, addresses[0], port, parsed.scheme, authority)
|
|
except CallbackURLValidationError:
|
|
raise
|
|
except (ValueError, UnicodeError, OSError) as exc:
|
|
raise CallbackURLValidationError("callback_url is malformed or DNS resolution failed") from exc
|