Fix SSRF in Python and Django

def _target(raw_url):
    """No validation at all: parse the URL and hand the host to the resolver."""
    parts = urlsplit(raw_url)
    host_header = parts.hostname if parts.port is None else f"{parts.hostname}:{parts.port}"
    path = (parts.path or "/") + (("?" + parts.query) if parts.query else "")
    return parts.hostname, (parts.port or default_port(parts.scheme)), host_header, path, f"{parts.scheme}://{parts.netloc}"

# Private, loopback, link-local and other non-public ranges in one table, so the policy
# is reviewable in one place.
_DENY_V4 = [ipaddress.ip_network(c) for c in (
    "0.0.0.0/8", "10.0.0.0/8", "100.64.0.0/10", "127.0.0.0/8", "169.254.0.0/16",
    "172.16.0.0/12", "192.0.0.0/24", "192.0.2.0/24", "192.88.99.0/24", "192.168.0.0/16",
    "198.18.0.0/15", "198.51.100.0/24", "203.0.113.0/24", "224.0.0.0/4", "240.0.0.0/4",
)]
_V6_GLOBAL_UNICAST = ipaddress.ip_network("2000::/3")
_V6_DOC = ipaddress.ip_network("2001:db8::/32")
_NAT64 = ipaddress.ip_network("64:ff9b::/96")


def parse_ip_literal(host):
    """inet_aton-style parse so decimal (2130706433), octal (0177.0.0.1) and hex
    (0x7f.0.0.1) literals are recognised as addresses rather than trusted as names."""
    inner = host[1:-1] if host.startswith("[") and host.endswith("]") else host
    try:
        return str(ipaddress.IPv6Address(inner))
    except ipaddress.AddressValueError:
        pass
    if inner != host:
        return None  # bracketed but not IPv6
    parts = host.split(".")
    if not 1 <= len(parts) <= 4:
        return None
    nums = []
    for part in parts:
        if re.fullmatch(r"0[xX][0-9a-fA-F]+", part):
            nums.append(int(part, 16))
        elif re.fullmatch(r"0[0-7]*", part):
            nums.append(int(part, 8))
        elif re.fullmatch(r"[1-9][0-9]*", part):
            nums.append(int(part))
        else:
            return None
    caps = [0xFFFFFFFF, 0xFFFFFF, 0xFFFF, 0xFF]
    if nums[-1] > caps[len(nums) - 1] or any(n > 0xFF for n in nums[:-1]):
        return None
    shifts = {1: [0], 2: [24, 0], 3: [24, 16, 0], 4: [24, 16, 8, 0]}[len(nums)]
    value = sum(n << s for n, s in zip(nums, shifts))
    return str(ipaddress.IPv4Address(value))


def assert_public_address(ip_str):
    ip = ipaddress.ip_address(ip_str)
    if isinstance(ip, ipaddress.IPv6Address):
        if ip.ipv4_mapped is not None:
            ip = ip.ipv4_mapped                       # ::ffff:a.b.c.d
        elif ip in _NAT64:
            ip = ipaddress.IPv4Address(int(ip) & 0xFFFFFFFF)  # 64:ff9b::a.b.c.d
    if isinstance(ip, ipaddress.IPv6Address):
        if ip not in _V6_GLOBAL_UNICAST or ip in _V6_DOC:
            raise BlockedRequestError("non-public IPv6 address")
        return
    if any(ip in net for net in _DENY_V4):
        raise BlockedRequestError("non-public IPv4 address")


def vet(raw_url, resolve):
    """Parse the URL, enforce the scheme and port allowlists, turn the host into a
    vetted address, and reject (not resolve) IP literals that are not public."""
    parts = urlsplit(raw_url)
    if parts.scheme not in ALLOWED_SCHEMES:
        raise BlockedRequestError("scheme not allowed")
    if parts.username or parts.password:
        raise BlockedRequestError("credentials in URL not allowed")
    if not parts.hostname:
        raise BlockedRequestError("missing host")
    try:
        port = parts.port or default_port(parts.scheme)
    except ValueError:
        raise BlockedRequestError("invalid port")
    if port not in ALLOWED_PORTS:
        raise BlockedRequestError("port not allowed")

    literal = parse_ip_literal(parts.hostname)
    addresses = [literal] if literal else resolve(parts.hostname)
    if not addresses:
        raise BlockedRequestError("host did not resolve")
    for ip in addresses:
        assert_public_address(ip)                     # reject the whole set if any is internal

    host_header = parts.hostname if parts.port is None else f"{parts.hostname}:{parts.port}"
    path = parts.path or "/"
    if parts.query:
        path += "?" + parts.query
    return addresses[0], port, host_header, path, f"{parts.scheme}://{parts.netloc}"

def fetch_url(raw_url, deps):
    """A 3xx response is a fresh destination decision, not permission to follow. Re-run
    the full policy on every hop and pin each connection to its own vetted address."""
    resolve = deps["resolve"]
    max_redirects = deps.get("max_redirects", MAX_REDIRECTS)
    target = raw_url
    for _ in range(max_redirects + 1):
        ip, port, host_header, path, _origin = vet(target, resolve)
        result = send_once(ip, port, host_header, path, deps["dial"],
                           max_bytes=deps["max_bytes"], timeout=deps["timeout"])
        if 300 <= result["status"] < 400 and result["location"]:
            target = urljoin(target, result["location"])  # relative to this hop's URL; validated again at the top
            continue
        return summarise(result)
    raise BlockedRequestError("too many redirects")

# A substring blocklist of "obviously internal" hosts. Looks reassuring; the address an
# attacker actually reaches is decided by DNS, which this never consults.
_BLOCKED_SUBSTRINGS = (
    "localhost", "127.0.0.1", "0.0.0.0", "169.254", "10.", "192.168", "172.16", "::1", "metadata", "internal",
)


def _screen(hostname):
    host = (hostname or "").lower()
    if any(bad in host for bad in _BLOCKED_SUBSTRINGS):
        raise BlockedRequestError("host is blocked")


def _target(raw_url):
    parts = urlsplit(raw_url)
    host_header = parts.hostname if parts.port is None else f"{parts.hostname}:{parts.port}"
    path = (parts.path or "/") + (("?" + parts.query) if parts.query else "")
    return parts.hostname, (parts.port or default_port(parts.scheme)), host_header, path, f"{parts.scheme}://{parts.netloc}"