# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-egress-policy-enforcement-point-fail-closed-on-tool-call-argument-resolution-rather-than-on-tool-name-allow.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-egress-policy-enforcement-point-fail-closed-on-tool-call-argument-resolution-rather-than-on-tool-name-allow.warrant.json
# Markdown SHA-256:   5ae448865eacb7a846620b3a1a0fec2e63301601260e133d225e319b8844c372
# Code AST SHA-256:   0a45eb02f471fa8601047bd36b056250af099a7a48da0d4832e609f79e606039
# Verified Classes:   EgressPolicyViolation
# Defensive Guards:   7 fail-closed invariants (zero external deps)

import ipaddress
from urllib.parse import urlparse


class EgressPolicyViolation(RuntimeError):
    """Raised when a resolved egress destination escapes policy."""


ALLOWED_TOOLS = {"fetch", "send_email", "write_file"}
ALLOWED_HOSTS = {"api.internal.example", "cdn.internal.example"}


def _resolve_destination(tool_name: str, resolved_args: dict) -> str:
    if tool_name == "fetch":
        return resolved_args["url"]
    if tool_name == "send_email":
        return resolved_args["to"]
    if tool_name == "write_file":
        return resolved_args["path"]
    raise EgressPolicyViolation(f"no destination resolver for tool {tool_name!r}")


def enforce_egress(tool_name: str, resolved_args: dict) -> None:
    # Pre-filter: name allowlist. Necessary, never sufficient.
    if tool_name not in ALLOWED_TOOLS:
        raise EgressPolicyViolation(f"tool {tool_name!r} not on allowlist")

    destination = _resolve_destination(tool_name, resolved_args)

    if tool_name == "fetch":
        parsed = urlparse(destination)
        host = parsed.hostname or ""
        # Block IP literals that resolve to link-local / loopback / private space.
        try:
            ip = ipaddress.ip_address(host)
            if ip.is_private or ip.is_loopback or ip.is_link_local:
                raise EgressPolicyViolation(
                    f"fetch destination {host!r} is a non-routable address"
                )
        except ValueError:
            pass  # hostname, not an IP literal
        if host not in ALLOWED_HOSTS:
            raise EgressPolicyViolation(
                f"fetch destination {host!r} not in allowed host set"
            )
        if parsed.scheme not in {"https"}:
            raise EgressPolicyViolation(
                f"fetch scheme {parsed.scheme!r} not permitted"
            )

    elif tool_name == "send_email":
        if not destination.endswith("@internal.example"):
            raise EgressPolicyViolation(
                f"email recipient {destination!r} is external"
            )

    elif tool_name == "write_file":
        import os
        canonical = os.path.realpath(destination)
        root = os.path.realpath("/srv/agent-workspace")
        if not canonical.startswith(root + os.sep):
            raise EgressPolicyViolation(
                f"write path {canonical!r} escapes workspace root {root!r}"
            )