diff --git a/app/security.py b/app/security.py index f8a9830..364dce5 100644 --- a/app/security.py +++ b/app/security.py @@ -1,85 +1,29 @@ import asyncio import ipaddress import socket -from collections.abc import Awaitable, Callable -from dataclasses import dataclass -from urllib.parse import SplitResult, urlsplit, urlunsplit +from urllib.parse import urlsplit -class TargetPolicyError(Exception): +class UnsafeDestination(ValueError): pass -@dataclass(frozen=True) -class ValidatedTarget: - url: str - hostname: str - port: int - addresses: tuple[str, ...] - - -Resolver = Callable[[str, int], Awaitable[list[tuple[object, ...]]]] - - -def redact_url(value: str) -> str: - parsed = urlsplit(value) - host = parsed.hostname or "invalid-host" - if ":" in host: - host = f"[{host}]" - port = f":{parsed.port}" if parsed.port else "" - return urlunsplit((parsed.scheme.lower(), f"{host}{port}", parsed.path or "/", "", "")) - - -def _public_address(value: str) -> bool: - address = ipaddress.ip_address(value) - return bool( - address.is_global - and not address.is_private - and not address.is_loopback - and not address.is_link_local - and not address.is_multicast - and not address.is_reserved - and not address.is_unspecified - ) - - -async def system_resolver(host: str, port: int) -> list[tuple[object, ...]]: - loop = asyncio.get_running_loop() - return await loop.getaddrinfo(host, port, type=socket.SOCK_STREAM, proto=socket.IPPROTO_TCP) - - -async def validate_target(url: str, resolver: Resolver = system_resolver) -> ValidatedTarget: +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: - parsed: SplitResult = urlsplit(url) - if parsed.scheme.lower() not in {"http", "https"}: - raise TargetPolicyError("only HTTP and HTTPS targets are allowed") - if parsed.username is not None or parsed.password is not None: - raise TargetPolicyError("target userinfo is not allowed") - if not parsed.hostname: - raise TargetPolicyError("target hostname is required") - hostname = parsed.hostname.rstrip(".").lower() - port = parsed.port or (443 if parsed.scheme.lower() == "https" else 80) - except ValueError as exc: - raise TargetPolicyError("target URL is invalid") from exc - - addresses: set[str] = set() - try: - literal = ipaddress.ip_address(hostname.split("%", 1)[0]) - addresses.add(str(literal)) + literal = ipaddress.ip_address(parsed.hostname) + addresses = [literal] except ValueError: + loop = asyncio.get_running_loop() try: - answers = await resolver(hostname, port) - for answer in answers: - sockaddr = answer[4] - if isinstance(sockaddr, tuple) and sockaddr: - addresses.add(str(ipaddress.ip_address(str(sockaddr[0]).split("%", 1)[0]))) - except (OSError, ValueError) as exc: - raise TargetPolicyError("target DNS resolution failed") from exc - - if not addresses: - raise TargetPolicyError("target DNS returned no addresses") - if len(addresses) > 32: - raise TargetPolicyError("target DNS returned too many addresses") - if not all(_public_address(address) for address in addresses): - raise TargetPolicyError("target resolves to a non-public address") - return ValidatedTarget(url=url, hostname=hostname, port=port, addresses=tuple(sorted(addresses))) + 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")