# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/amid-what-egress-and-name-resolution-guards-must-an-instance-validate-user-supplied-urls-to-prevent-server-side-request-forgery-and-metadata-exfiltration.md
# Remnant Warrant:    https://corpusfetch.com/c/amid-what-egress-and-name-resolution-guards-must-an-instance-validate-user-supplied-urls-to-prevent-server-side-request-forgery-and-metadata-exfiltration.warrant.json
# Markdown SHA-256:   d48557ebf227726bc7635441ae575559d7b6b53ed0418359233945a1d3b5f7bd
# Code AST SHA-256:   1858b8bfc0ae9a85b742deeba922ea495133449c181cdc2398d74e6ac66acfd1
# Verified Classes:   _NoRedirectHandler
# Defensive Guards:   9 fail-closed invariants (zero external deps)

import ipaddress
import socket
import urllib.parse
import urllib.request
from typing import Sequence


def verify_public_egress_url(
    raw_url: str,
    allowed_ports: Sequence[int] = (80, 443),
) -> tuple[str, str, int, list[str]]:
    '''
    Validates that a URL uses http/https on an allowed port and resolves
    exclusively to globally routable public IP addresses (blocking localhost,
    RFC 1918 private networks, and 169.254.169.254 cloud metadata endpoints).
    '''
    parsed = urllib.parse.urlsplit(raw_url.strip())

    # Gate 1: Strict scheme check (blocks file://, gopher://, ftp://)
    scheme = parsed.scheme.lower()
    if scheme not in ("http", "https"):
        raise PermissionError(f"Blocked disallowed URL scheme: {scheme!r}")

    # Gate 2: Reject embedded credentials (user:pass@host) and missing hostnames
    if parsed.username or parsed.password:
        raise PermissionError("Blocked URL containing embedded userinfo credentials")
    hostname = parsed.hostname
    if not hostname:
        raise ValueError(f"Malformed URL missing hostname: {raw_url!r}")

    # Gate 3: Strict port allowlist
    port = parsed.port or (443 if scheme == "https" else 80)
    if port not in allowed_ports:
        raise PermissionError(f"Blocked disallowed egress port: {port}")

    # Gate 4: Resolve DNS and inspect every candidate IPv4 / IPv6 address
    try:
        addr_info = socket.getaddrinfo(hostname, port, type=socket.SOCK_STREAM)
    except socket.gaierror as exc:
        raise ConnectionError(f"DNS resolution failed for {hostname!r}") from exc

    verified_ips: list[str] = []
    for family, _, _, _, sockaddr in addr_info:
        raw_ip = sockaddr[0]
        ip_obj = ipaddress.ip_address(raw_ip)

        # Unwrap IPv4-mapped IPv6 addresses (e.g., ::ffff:127.0.0.1 -> 127.0.0.1)
        if isinstance(ip_obj, ipaddress.IPv6Address) and ip_obj.ipv4_mapped is not None:
            ip_obj = ip_obj.ipv4_mapped

        # Must be strictly global and free of all internal/special flags
        if (
            not ip_obj.is_global
            or ip_obj.is_private
            or ip_obj.is_loopback
            or ip_obj.is_link_local
            or ip_obj.is_multicast
            or ip_obj.is_reserved
            or ip_obj.is_unspecified
        ):
            raise PermissionError(
                f"SSRF blocked: host {hostname!r} resolved to non-public IP {ip_obj}"
            )
        verified_ips.append(str(ip_obj))

    if not verified_ips:
        raise PermissionError(f"No valid public IP addresses resolved for {hostname!r}")

    return parsed.geturl(), hostname, port, sorted(set(verified_ips))


class _NoRedirectHandler(urllib.request.HTTPRedirectHandler):
    '''Prevents urllib from silently following 301/302/307/308 redirects to internal IPs.'''

    def redirect_request(self, req, fp, code, msg, headers, newurl):
        raise PermissionError(
            f"Blocked automatic HTTP {code} redirect to {newurl!r}. "
            "Re-validate redirect target through verify_public_egress_url first."
        )


def fetch_public_url_bounded(
    raw_url: str,
    *,
    timeout_sec: float = 10.0,
    max_bytes: int = 2 * 1024 * 1024,
) -> bytes:
    '''
    Fetches an external URL only after verifying all resolved IPs are public,
    with automatic redirects disabled and a strict response byte ceiling.
    '''
    clean_url, _, _, _ = verify_public_egress_url(raw_url)
    opener = urllib.request.build_opener(_NoRedirectHandler)
    req = urllib.request.Request(clean_url, headers={"User-Agent": "CorpusFetch-SafeClient/1.0"})

    with opener.open(req, timeout=timeout_sec) as resp:
        data = resp.read(max_bytes + 1)
        if len(data) > max_bytes:
            raise RuntimeError(f"Response from {clean_url} exceeded {max_bytes} byte limit")
        return data