87 lines
2.9 KiB
Python
87 lines
2.9 KiB
Python
"""URL safety helpers — prevent SSRF from user-supplied fetch URLs
|
|
(calendar subscriptions, weather API hosts)."""
|
|
import ipaddress
|
|
import socket
|
|
from urllib.parse import urlparse
|
|
|
|
ALLOWED_SCHEMES = ("http", "https")
|
|
|
|
# Hosts that are always fine to reach (QWeather SaaS / common cloud APIs).
|
|
# Everything else is resolved and must be a public address.
|
|
_ALWAYS_ALLOWED = {
|
|
"devapi.qweather.com",
|
|
"geoapi.qweather.com",
|
|
"qweather.com",
|
|
}
|
|
|
|
|
|
def _host_is_private(hostname: str) -> bool:
|
|
"""Resolve a hostname and return True if ANY address is non-public."""
|
|
try:
|
|
infos = socket.getaddrinfo(hostname, None)
|
|
except Exception:
|
|
# Unresolvable — treat as unsafe (fail closed for fetches).
|
|
return True
|
|
for info in infos:
|
|
ip = ipaddress.ip_address(info[4][0])
|
|
if (ip.is_private or ip.is_loopback or ip.is_link_local
|
|
or ip.is_multicast or ip.is_reserved or ip.is_unspecified):
|
|
return True
|
|
return False
|
|
|
|
|
|
def is_safe_public_url(url: str) -> bool:
|
|
"""True if the URL is http(s) and resolves to public addresses only."""
|
|
if not url or not isinstance(url, str):
|
|
return False
|
|
try:
|
|
p = urlparse(url.strip())
|
|
except Exception:
|
|
return False
|
|
if p.scheme not in ALLOWED_SCHEMES or not p.hostname:
|
|
return False
|
|
host = p.hostname.lower().rstrip(".")
|
|
if host in _ALWAYS_ALLOWED or host.endswith(".qweather.com"):
|
|
return True
|
|
if p.port and p.port not in (80, 443):
|
|
return False
|
|
return not _host_is_private(host)
|
|
|
|
|
|
def is_safe_https_host(host: str) -> bool:
|
|
"""True if the value is a bare https hostname (no scheme/path) and public."""
|
|
if not host or not isinstance(host, str):
|
|
return False
|
|
host = host.strip().lower().rstrip(".")
|
|
if not host or "://" in host or "/" in host or " " in host:
|
|
return False
|
|
if host in _ALWAYS_ALLOWED or host.endswith(".qweather.com"):
|
|
return True
|
|
return not _host_is_private(host)
|
|
|
|
|
|
def safe_http_get(url: str, timeout: float = 15, max_redirects: int = 3):
|
|
"""GET a URL, re-validating safety on every redirect hop.
|
|
|
|
Returns the final `requests.Response` (status 200) or None if any hop is
|
|
unsafe/unresolvable or the redirect chain exceeds max_redirects.
|
|
Raises requests.HTTPError for non-2xx final responses.
|
|
"""
|
|
import requests
|
|
from urllib.parse import urljoin
|
|
|
|
current = (url or "").strip()
|
|
for _ in range(max_redirects + 1):
|
|
if not is_safe_public_url(current):
|
|
return None
|
|
resp = requests.get(current, timeout=timeout, allow_redirects=False)
|
|
if resp.status_code in (301, 302, 303, 307, 308):
|
|
location = resp.headers.get("Location", "")
|
|
if not location:
|
|
return None
|
|
current = urljoin(current, location)
|
|
continue
|
|
resp.raise_for_status()
|
|
return resp
|
|
return None
|