# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-should-an-instance-reconcile-egress-allowlists-with-argument-schemas-when-a-tool-s-effective-destination-is-derived-from-its-arguments.md
# Remnant Warrant:    https://corpusfetch.com/c/how-should-an-instance-reconcile-egress-allowlists-with-argument-schemas-when-a-tool-s-effective-destination-is-derived-from-its-arguments.warrant.json
# Markdown SHA-256:   cbda2ba490a8d2d0c0f85d7e656a569568edc0c2c321027236a830e579cd7776
# Code AST SHA-256:   3e474953c7bc7ab9b28c96c1fc4072c3b157198c6025f0de11df33cfd944dc64
# Verified Classes:   EgressInvariantViolation, EgressDecision
# Defensive Guards:   5 fail-closed invariants (zero external deps)

import ipaddress
import urllib.parse
from dataclasses import dataclass


class EgressInvariantViolation(RuntimeError):
    """Raised when the host checked against the allowlist differs from the host the transport will connect to."""


@dataclass(frozen=True)
class EgressDecision:
    canonical_host: str
    allowed: bool


ALLOWED_HOSTS = frozenset({"api.internal.example.com"})
METADATA_PREFIXES = ("169.254.", "127.", "10.", "192.168.", "::1")


def canonicalize_host(raw_url: str) -> str:
    parts = urllib.parse.urlsplit(raw_url)
    host = parts.hostname  # strips userinfo, lowercases, handles brackets
    if host is None:
        raise EgressInvariantViolation(f"no host component in {raw_url!r}")
    host = host.rstrip(".").lower()
    if host.startswith("[") and host.endswith("]"):
        host = host[1:-1]
    return host


def is_metadata_or_private(host: str) -> bool:
    try:
        addr = ipaddress.ip_address(host)
    except ValueError:
        return False
    return addr.is_private or addr.is_loopback or addr.is_link_local


def enforce_egress(raw_url: str, *, transport_host: str | None = None) -> EgressDecision:
    """
    Enforce at the connection boundary, not at schema validation.
    `transport_host` is the host the client library actually resolved;
    if it is supplied and differs from our canonical parse, the invariant
    has been broken somewhere and we refuse rather than guess.
    """
    canonical = canonicalize_host(raw_url)

    if transport_host is not None and transport_host.rstrip(".").lower() != canonical:
        raise EgressInvariantViolation(
            f"normalization divergence: policy saw {canonical!r}, transport will connect to {transport_host!r}"
        )

    if is_metadata_or_private(canonical):
        raise EgressInvariantViolation(f"refusing private/metadata destination {canonical!r}")

    if canonical not in ALLOWED_HOSTS:
        return EgressDecision(canonical_host=canonical, allowed=False)

    return EgressDecision(canonical_host=canonical, allowed=True)


def fetch_resource(url: str, *, follow_redirects: bool = False) -> EgressDecision:
    decision = enforce_egress(url)
    if not decision.allowed:
        raise EgressInvariantViolation(f"egress denied for {decision.canonical_host!r}")
    if follow_redirects:
        # Redirects must be re-checked per hop; a single pre-dispatch check is unsound.
        raise EgressInvariantViolation("redirect-following requires per-hop egress re-evaluation")
    return decision