kopia lustrzana https://github.com/tsileo/little-boxes
67 wiersze
1.7 KiB
Python
67 wiersze
1.7 KiB
Python
import ipaddress
|
|
import logging
|
|
import socket
|
|
from typing import Dict
|
|
from urllib.parse import urlparse
|
|
|
|
from .errors import Error
|
|
from .errors import ServerError
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
_CACHE: Dict[str, bool] = {}
|
|
|
|
|
|
class InvalidURLError(ServerError):
|
|
pass
|
|
|
|
|
|
class URLLookupFailedError(Error):
|
|
pass
|
|
|
|
|
|
def is_url_valid(url: str, debug: bool = False) -> bool:
|
|
parsed = urlparse(url)
|
|
if parsed.scheme not in ["http", "https"]:
|
|
return False
|
|
|
|
# XXX in debug mode, we want to allow requests to localhost to test the federation with local instances
|
|
if debug: # pragma: no cover
|
|
return True
|
|
|
|
if parsed.hostname in ["localhost"]:
|
|
return False
|
|
|
|
if _CACHE.get(parsed.hostname, False):
|
|
return True
|
|
|
|
try:
|
|
ip_address = ipaddress.ip_address(parsed.hostname)
|
|
except ValueError:
|
|
try:
|
|
ip_address = socket.getaddrinfo(parsed.hostname, parsed.port or 80)[0][4][0]
|
|
logger.debug(f"dns lookup: {parsed.hostname} -> {ip_address}")
|
|
except socket.gaierror:
|
|
logger.exception(f"failed to lookup url {url}")
|
|
_CACHE[parsed.hostname] = False
|
|
raise URLLookupFailedError(f"failed to lookup url {url}")
|
|
|
|
logger.debug(f"{ip_address}")
|
|
|
|
if ipaddress.ip_address(ip_address).is_private:
|
|
logger.info(f"rejecting private URL {url}")
|
|
_CACHE[parsed.hostname] = False
|
|
return False
|
|
|
|
_CACHE[parsed.hostname] = True
|
|
return True
|
|
|
|
|
|
def check_url(url: str, debug: bool = False) -> None:
|
|
logger.debug(f"check_url {url} debug={debug}")
|
|
if not is_url_valid(url, debug=debug):
|
|
raise InvalidURLError(f'"{url}" is invalid')
|
|
|
|
return None
|