63 lines
2.0 KiB
Python
63 lines
2.0 KiB
Python
import asyncio
|
|
import ipaddress
|
|
import socket
|
|
from collections.abc import Awaitable, Callable
|
|
from urllib.parse import urlsplit
|
|
|
|
|
|
class UnsafePushEndpoint(ValueError):
|
|
"""Raised when a Web Push endpoint could reach a non-public service."""
|
|
|
|
|
|
Resolver = Callable[[str, int], Awaitable[list[str]]]
|
|
|
|
|
|
async def _resolve(host: str, port: int) -> list[str]:
|
|
loop = asyncio.get_running_loop()
|
|
records = await loop.getaddrinfo(
|
|
host,
|
|
port,
|
|
family=socket.AF_UNSPEC,
|
|
type=socket.SOCK_STREAM,
|
|
)
|
|
return sorted({record[4][0] for record in records})
|
|
|
|
|
|
async def validate_public_push_endpoint(
|
|
endpoint: str,
|
|
*,
|
|
resolver: Resolver = _resolve,
|
|
timeout_seconds: float = 2.0,
|
|
) -> str:
|
|
"""Return a canonical public HTTPS push endpoint or fail closed."""
|
|
parsed = urlsplit(endpoint)
|
|
try:
|
|
port = parsed.port
|
|
except ValueError as error:
|
|
raise UnsafePushEndpoint(
|
|
"Endpoint must be a canonical public Web Push service URL"
|
|
) from error
|
|
if (
|
|
parsed.scheme != "https"
|
|
or not parsed.hostname
|
|
or parsed.username is not None
|
|
or parsed.password is not None
|
|
or parsed.fragment
|
|
or port not in {None, 443}
|
|
):
|
|
raise UnsafePushEndpoint("Endpoint must be a canonical public Web Push service URL")
|
|
try:
|
|
addresses = await asyncio.wait_for(
|
|
resolver(parsed.hostname, port or 443),
|
|
timeout=max(0.01, timeout_seconds),
|
|
)
|
|
except (OSError, asyncio.TimeoutError) as error:
|
|
raise UnsafePushEndpoint("Endpoint must resolve to a public Web Push service") from error
|
|
try:
|
|
public = addresses and all(ipaddress.ip_address(address).is_global for address in addresses)
|
|
except ValueError as error:
|
|
raise UnsafePushEndpoint("Endpoint must resolve to a public Web Push service") from error
|
|
if not public:
|
|
raise UnsafePushEndpoint("Endpoint must resolve to a public Web Push service")
|
|
return endpoint
|