diff --git a/app/security.py b/app/security.py index 0b3a61d..900ce2a 100644 --- a/app/security.py +++ b/app/security.py @@ -2,7 +2,7 @@ import asyncio import ipaddress import socket from collections.abc import Awaitable, Callable -from urllib.parse import urlsplit +from urllib.parse import urlsplit, urlunsplit Resolver = Callable[[str, int], Awaitable[list[str]]] @@ -12,27 +12,55 @@ class UnsafeTarget(ValueError): async def system_resolver(host: str, port: int) -> list[str]: - loop = asyncio.get_running_loop() - records = await loop.getaddrinfo(host, port, type=socket.SOCK_STREAM) - return sorted({record[4][0] for record in records}) + def resolve() -> list[str]: + records = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM) + return sorted({record[4][0] for record in records}) + + try: + return await asyncio.to_thread(resolve) + except socket.gaierror as exc: + raise OSError(f"DNS resolution failed: {exc}") from exc + + +def _require_global(address: str) -> None: + try: + ip = ipaddress.ip_address(address) + except ValueError as exc: + raise UnsafeTarget("DNS returned an invalid address") from exc + if not ip.is_global: + raise UnsafeTarget("target resolves to a non-public address") async def validate_target(url: str, resolver: Resolver = system_resolver) -> None: parsed = urlsplit(url) - if parsed.scheme not in {"http", "https"} or not parsed.hostname: - raise UnsafeTarget("unsupported_target") - if parsed.username or parsed.password: - raise UnsafeTarget("credentials_not_allowed") - port = parsed.port or (443 if parsed.scheme == "https" else 80) + if parsed.scheme not in {"http", "https"}: + raise UnsafeTarget("only HTTP(S) targets are allowed") + if parsed.username is not None or parsed.password is not None: + raise UnsafeTarget("URL credentials are not allowed") + if not parsed.hostname: + raise UnsafeTarget("target has no hostname") try: - addresses = await resolver(parsed.hostname, port) - except (OSError, socket.gaierror) as exc: - raise UnsafeTarget("dns_resolution_failed") from exc - if not addresses: - raise UnsafeTarget("dns_resolution_failed") + literal = ipaddress.ip_address(parsed.hostname) + except ValueError: + addresses = await resolver(parsed.hostname, parsed.port or (443 if parsed.scheme == "https" else 80)) + if not addresses: + raise OSError("DNS resolution returned no addresses") + for address in addresses: + _require_global(address) + else: + _require_global(str(literal)) + + +def redact_url(url: str) -> str: + """Retain scheme/host/port/path, never credentials, query, or fragment.""" + parsed = urlsplit(url) + host = parsed.hostname or "" + if ":" in host and not host.startswith("["): + host = f"[{host}]" + netloc = host try: - safe = all(ipaddress.ip_address(address).is_global for address in addresses) - except ValueError as exc: - raise UnsafeTarget("dns_resolution_failed") from exc - if not safe: - raise UnsafeTarget("non_public_target") + if parsed.port is not None: + netloc = f"{host}:{parsed.port}" + except ValueError: + netloc = host + return urlunsplit((parsed.scheme, netloc, parsed.path, "", ""))