# Natural Language Toolkit: Centralized I/O security sentinel
#
# Copyright (C) 2001-2026 NLTK Project
# Author: Eric Kafe <kafe.eric@gmail.com>
# URL: <https://www.nltk.org/>
# For license information, see LICENSE.TXT
#
"""Centralized I/O security sentinel for NLTK."""

"""Centralized I/O security sentinel for NLTK."""
import builtins
import http.client
import ipaddress
import os
import socket
import stat
import sys
import tempfile
import urllib.request
import warnings
import zipfile
from functools import lru_cache
from pathlib import Path
from urllib.parse import unquote, urlparse

# Security Enforcement Toggle
# ENFORCE = False
ENFORCE = True

# When a proxy is configured, the proxy -- not NLTK -- resolves the hostname and
# performs the egress, so NLTK cannot pin the validated destination IP and its
# SSRF filter no longer governs where the request actually lands (CWE-918,
# GHSA-6ww7). Proxied fetches are therefore refused under ``ENFORCE`` unless the
# operator explicitly opts in here (or via ``NLTK_ALLOW_PROXIED_URLOPEN``),
# asserting that the proxy itself is trusted to be SSRF-safe.
ALLOW_PROXIED_FETCH = False

_ALLOWED_ROOTS_CACHE = None
_LAST_DATA_PATHS = None


def is_private_dir(path):
    """Return True if *path* is a directory safe to trust as a data root: another
    *unprivileged local user* cannot plant files in it.

    This is the test that distinguishes a shared, world-writable temp directory
    (Linux ``/tmp``, mode ``1777``) -- which a local attacker could use to plant
    trusted-looking data (CWE-377/CWE-378) -- from a *private* per-user temp
    directory (macOS ``$TMPDIR`` ``/var/folders/...`` mode ``0700``, Windows
    ``%TEMP%`` under the ACL-protected user profile), which is safe.

    Cross-platform:
      * POSIX: the directory must be owned by the current user (or root) and be
        neither group- nor world-writable (unless the writable group/other bit
        is paired with the sticky bit *and* the dir is user-owned -- but we keep
        it strict and simply reject any group/other-writable dir).
      * Windows: ``st_uid``/mode are not meaningful; the per-user temp/profile is
        ACL-protected, so a plain "exists and is a directory" check is used.
    """
    try:
        st = os.stat(path)
    except OSError:
        return False
    if not stat.S_ISDIR(st.st_mode):
        return False
    if os.name != "posix":
        # Windows / other: rely on the per-user profile ACLs (no POSIX mode).
        return True
    # Must be owned by us (or by root, e.g. a system-wide /usr/share dir).
    if st.st_uid not in (os.getuid(), 0):
        return False
    # Reject group- or world-writable directories: those let another account
    # write into (or, without the sticky bit, replace) the directory.
    if st.st_mode & (stat.S_IWGRP | stat.S_IWOTH):
        return False
    return True


def _get_allowed_roots():
    """Dynamically determines allowed directories based on NLTK data paths."""
    global _ALLOWED_ROOTS_CACHE, _LAST_DATA_PATHS

    current_paths = []
    if "nltk.data" in sys.modules:
        # Accessing nltk.data.path via sys.modules to avoid top-level circularity
        current_paths = list(getattr(sys.modules["nltk.data"], "path", []))

    env_paths = os.environ.get("NLTK_DATA", "")
    current_state = (current_paths, env_paths)

    if _ALLOWED_ROOTS_CACHE is not None and _LAST_DATA_PATHS == current_state:
        return _ALLOWED_ROOTS_CACHE

    roots = set()
    for p in current_paths + env_paths.split(os.pathsep):
        if p:
            try:
                # Handle both string paths and PathPointer objects
                raw_p = p.path if hasattr(p, "path") else p
                roots.add(Path(str(raw_p)).expanduser().resolve())
            except (OSError, ValueError, RuntimeError):
                continue

    candidate_locs = ["~/nltk_data", "/usr/share/nltk_data"]

    for loc in candidate_locs:
        try:
            p = Path(loc).expanduser().resolve()
            if p.exists():
                roots.add(p)
        except (OSError, ValueError, RuntimeError):
            continue

    # The system temp directory is trusted ONLY when it is private to the current
    # user.  A *shared, world-writable* temp dir (Linux ``/tmp``, mode ``1777``)
    # must not be an allowed root: a local attacker could plant trusted-looking
    # data there (CWE-377/CWE-378, GHSA-p4rw follow-up).  A *private* per-user
    # temp dir (macOS ``/var/folders/...`` mode ``0700``, Windows ``%TEMP%`` under
    # the user profile, or a private Linux ``$TMPDIR``) is safe and is where
    # NLTK / its test-suite legitimately stage temporary corpora, so it is kept.
    try:
        tmp = Path(tempfile.gettempdir()).resolve()
        if is_private_dir(tmp):
            roots.add(tmp)
    except (OSError, ValueError, RuntimeError):
        pass

    _ALLOWED_ROOTS_CACHE = roots
    _LAST_DATA_PATHS = current_state
    return roots


def validate_path(path_input, context="NLTK", required_root=None):
    """
    Ensures file access is restricted to allowed data directories.

    :param path_input: The path to validate.
    :param context: Diagnostic context for warnings/errors.
    :param required_root: If provided, enforces that the path is strictly
                          within this specific directory (scoped sandbox).
    """
    if isinstance(path_input, int) or not path_input or not str(path_input).strip():
        return
    try:
        raw = path_input.path if hasattr(path_input, "path") else str(path_input)

        if "://" in raw:
            parsed = urlparse(raw)
            if parsed.scheme in ("http", "https", "ftp"):
                return
            if parsed.scheme == "file":
                raw = unquote(parsed.path)

        # Resolve path to catch symlink escapes
        try:
            target = Path(raw).resolve()
        except (OSError, ValueError):
            # Fallback for virtual paths inside ZIPs (e.g. corpora/foo.zip/file.txt)
            lower_raw = raw.lower()
            if ".zip" in lower_raw:
                zip_idx = lower_raw.find(".zip") + 4
                target = Path(raw[:zip_idx]).resolve()
            else:
                target = Path(raw)

        # LAYER 1: Scoped Sandbox (PR #3528 Integration)
        # This resolves both target and root to block symlink-based escapes.
        if required_root:
            root_raw = (
                required_root.path
                if hasattr(required_root, "path")
                else str(required_root)
            )
            scoped_root = Path(root_raw).resolve()
            if not (target == scoped_root or target.is_relative_to(scoped_root)):
                # Raise ValueError to match NLTK's historical CorpusReader error type
                raise ValueError(
                    f"Security Violation [{context}]: Path {target} escapes root {scoped_root}"
                )

        # LAYER 2: Global NLTK_DATA Sandbox
        allowed_roots = _get_allowed_roots()
        if any(target == root or target.is_relative_to(root) for root in allowed_roots):
            return

        # CWD Fallback (Explicit Opt-In for ENFORCE mode)
        try:
            cwd = Path(os.getcwd()).resolve()
            if target == cwd or target.is_relative_to(cwd):
                if any(cwd == root for root in allowed_roots):
                    return
                msg = (
                    f"Security Violation [{context}]: CWD access restricted in ENFORCE mode. "
                    "Authorize via: nltk.data.path.append('.')"
                )
                if ENFORCE:
                    raise PermissionError(msg)
                else:
                    warnings.warn(
                        f"Security Warning [{context}]: Path {target} allowed via CWD.",
                        RuntimeWarning,
                        stacklevel=3,
                    )
                    return
        except (OSError, ValueError):
            pass

        msg = f"Security Violation [{context}]: Unauthorized path {target}"
        if ENFORCE:
            raise PermissionError(msg)
        else:
            warnings.warn(msg, RuntimeWarning, stacklevel=3)
    except (PermissionError, ValueError):
        raise
    except Exception:
        if ENFORCE:
            raise


def _zip_member_is_unsafe(name_str):
    """True if a ZIP member is written somewhere other than where it is validated.

    ``zipfile.ZipFile.extract`` sanitises a member name by *dropping* the drive
    and every empty / ``.`` / ``..`` component while keeping the rest, whereas
    ``Path.resolve`` collapses a ``..`` against its *preceding* component.  For a
    member such as ``a/../b/x`` the two disagree: it is validated as ``<root>/b/x``
    but written to ``<root>/a/b/x``, which can escape through a pre-existing
    symlink at ``<root>/a/b`` that the collapsed validation path never visits.

    The mismatch only ever arises from absolute / drive-qualified / ``..`` members
    -- exactly the shapes a legitimate archive never uses -- so rather than keep
    two different normalizations in sync we reject them outright.  This both
    closes the validate/extract gap and is the proactive block the hardened
    extractor promises (CWE-22 / CWE-59).
    """
    # Normalize every separator zipfile treats as such on this platform to "/".
    normalized = name_str.replace("\\", "/") if os.path.altsep else name_str
    if os.path.splitdrive(name_str)[0] or normalized.startswith("/"):
        return True
    return os.path.pardir in normalized.split("/")


def validate_zip_archive(
    zip_obj_or_path, target_root, specific_member=None, context="ZipAudit"
):
    """Enhanced Zip-Slip protection using Pathlib for cross-platform safety."""
    try:
        target = Path(target_root).resolve()

        def _audit(zf):
            members = (
                [specific_member] if specific_member is not None else zf.namelist()
            )
            for name in members:
                name_str = name.filename if hasattr(name, "filename") else str(name)
                if "\0" in name_str:
                    raise ValueError(f"Null byte in ZIP member: {name_str}")

                # ``resolve()`` follows symlinks, catching escapes through a
                # pre-existing symlinked subpath. The extra component check
                # rejects absolute / ``..`` members, whose write target diverges
                # from this resolved path (CWE-22 / CWE-59).
                member_path = (target / name_str).resolve()
                if _zip_member_is_unsafe(name_str) or not (
                    member_path == target or member_path.is_relative_to(target)
                ):
                    msg = f"Security Violation [{context}]: Traversal member '{name_str}' detected."
                    if ENFORCE:
                        raise PermissionError(msg)
                    else:
                        warnings.warn(msg, RuntimeWarning, stacklevel=3)

        if isinstance(zip_obj_or_path, zipfile.ZipFile):
            _audit(zip_obj_or_path)
        else:
            with zipfile.ZipFile(zip_obj_or_path, "r") as zf:
                _audit(zf)
    except (PermissionError, ValueError):
        raise
    except (OSError, zipfile.BadZipFile):
        if ENFORCE:
            raise PermissionError("Zip validation failed")


@lru_cache(maxsize=256)
def _resolve_hostname(hostname):
    """Cached hostname resolution for the early SSRF pre-check.

    Note: the cache alone does NOT prevent DNS rebinding, because the connection
    layer re-resolves the hostname independently. The actual rebinding
    protection is the connect-time IP pinning in ``_SafeHTTPConnection`` /
    ``_SafeHTTPSConnection``.
    """
    try:
        return socket.getaddrinfo(hostname, None, proto=socket.IPPROTO_TCP)
    except (OSError, ValueError):
        return []


# IPv6->IPv4 transition prefixes that have no dedicated stdlib accessor.
_NAT64_WELL_KNOWN = ipaddress.ip_network("64:ff9b::/96")
_IPV4_COMPATIBLE = ipaddress.ip_network("::/96")


def _embedded_ipv4(ip):
    """The embedded IPv4 for IPv6 forms that *are* an IPv4 address, else None.

    Covers IPv4-mapped (``::ffff:0:0/96``), IPv4-compatible (``::/96``) and the
    NAT64 well-known prefix (``64:ff9b::/96``). For these the IPv6 wrapper has no
    independent routable meaning, so the embedded IPv4 is what gets reached.
    """
    if not isinstance(ip, ipaddress.IPv6Address):
        return None
    mapped = ip.ipv4_mapped
    if mapped is not None:
        return mapped
    if ip in _NAT64_WELL_KNOWN or ip in _IPV4_COMPATIBLE:
        return ipaddress.IPv4Address(ip.packed[-4:])
    return None


def _tunneled_ipv4s(ip):
    """IPv4 addresses tunneled by routable IPv6 wrappers (6to4 / Teredo)."""
    if not isinstance(ip, ipaddress.IPv6Address):
        return
    sixtofour = ip.sixtofour
    if sixtofour is not None:
        yield sixtofour
    teredo = ip.teredo
    if teredo is not None:
        yield from teredo  # (Teredo server, Teredo client)


def _ip_is_forbidden(ip):
    """Return True if the SSRF filter must refuse to connect to ``ip``.

    Policy (defense in depth): only *globally routable* addresses are allowed;
    anything that is not global -- loopback, link-local, private, carrier-grade
    NAT (100.64.0.0/10), reserved, unspecified (``0.0.0.0`` / ``::``),
    documentation ranges, etc. -- is forbidden. This generalises the previous
    explicit ``loopback / link-local / multicast / private`` list and is a strict
    superset of it. Multicast is still rejected explicitly because some CPython
    versions classify multicast addresses as ``is_global``.

    IPv6 addresses that embed an IPv4 address are evaluated by that embedded IPv4,
    not by the wrapper: the stdlib classifies the wrappers (IPv4-mapped,
    IPv4-compatible, NAT64 ``64:ff9b::/96``, 6to4 ``2002::/16``, Teredo
    ``2001:0::/32``) as globally routable, so a forbidden internal IPv4 (loopback,
    the link-local cloud-metadata address, ...) could otherwise be smuggled past
    the check and then routed to that IPv4 by a NAT64/6to4/Teredo gateway
    (CWE-918). This extends the previous IPv4-mapped-only unwrap.
    """
    embedded = _embedded_ipv4(ip)
    if embedded is not None:
        ip = embedded
    for tunneled in _tunneled_ipv4s(ip):
        if tunneled.is_multicast or not tunneled.is_global:
            return True
    return ip.is_multicast or not ip.is_global


def validate_network_url(url_input, context="NetworkIO"):
    """Hardened URL validation with SSRF protection."""
    if not url_input or not str(url_input).strip():
        return
    try:
        parsed = urlparse(str(url_input))

        if parsed.scheme == "file":
            file_path = unquote(parsed.path)
            netloc = parsed.netloc

            # Only local file:// URIs are allowed.
            # Reject remote/UNC-style authorities so validation matches actual access.
            if netloc not in ("", "localhost"):
                raise OSError(
                    f"Security Violation [{context}.file_scheme]: "
                    f"Non-local file URI authority not allowed: {netloc!r}"
                )

            # Windows file:// URIs arrive like /C:/path/to/file
            # Convert them to a native absolute path before validation.
            if (
                os.name == "nt"
                and len(file_path) >= 3
                and file_path[0] == "/"
                and file_path[2] == ":"
            ):
                file_path = file_path[1:]

            validate_path(file_path, context=f"{context}.file_scheme")
            return

        if parsed.scheme not in ("http", "https"):
            msg = (
                f"Security Violation [{context}]: Unsupported scheme '{parsed.scheme}'."
            )
            if ENFORCE:
                raise PermissionError(msg)
            else:
                warnings.warn(msg, RuntimeWarning, stacklevel=3)
            return

        for result in _resolve_hostname(parsed.hostname or ""):
            ip = ipaddress.ip_address(result[4][0])
            if _ip_is_forbidden(ip):
                msg = f"Security Violation [{context}]: SSRF attempt to restricted IP {ip}"
                if ENFORCE:
                    raise PermissionError(msg)
                else:
                    warnings.warn(msg, RuntimeWarning, stacklevel=3)
    except (PermissionError, ValueError):
        raise
    except Exception:
        if ENFORCE:
            raise


class _ValidatingRedirectHandler(urllib.request.HTTPRedirectHandler):
    """Ensures that every step of a redirect chain is re-validated against SSRF."""

    def redirect_request(self, req, fp, code, msg, headers, newurl):
        validate_network_url(newurl, context="NetworkRedirect")
        return super().redirect_request(req, fp, code, msg, headers, newurl)


def _resolve_and_validate_host(host, port):
    """
    Resolve ``host`` once and SSRF-validate *every* address it resolves to.

    Returns the resolved ``getaddrinfo`` records so the caller can connect to a
    **pinned** numeric address. Because validation and the subsequent connection
    observe the same resolution (the connection is made to the numeric IP, which
    triggers no further name lookup), this closes the DNS-rebinding TOCTOU where
    a hostname resolves to a public IP during validation and to an internal /
    loopback IP during the actual connect.
    """
    try:
        addrinfo = socket.getaddrinfo(host, port, proto=socket.IPPROTO_TCP)
    except (OSError, ValueError):
        return []
    for res in addrinfo:
        try:
            ip = ipaddress.ip_address(res[4][0])
        except ValueError:
            continue
        if _ip_is_forbidden(ip):
            msg = f"Security Violation [pathsec.urlopen]: SSRF attempt to restricted IP {ip}"
            if ENFORCE:
                raise PermissionError(msg)
            warnings.warn(msg, RuntimeWarning, stacklevel=2)
    return addrinfo


def _pinned_connection(host, port, timeout, source_address):
    """Open a socket to ``host``/``port`` over an SSRF-validated, pinned address.

    Every address ``host`` resolves to is validated together, then the socket is
    opened to those **numeric** addresses, so no second (unvalidated) name lookup
    happens at connect time -- this closes the DNS-rebinding TOCTOU.  All
    validated addresses are tried in order, preserving urllib/socket's normal
    dual-stack / multi-A fallback.  If nothing resolves to a validated address we
    fail closed: we never fall back to connecting by the raw hostname, which
    would re-resolve unvalidated and reopen the rebinding hole.
    """
    addrinfo = _resolve_and_validate_host(host, port)
    if not addrinfo:
        # Fail closed: never fall back to connecting by the raw hostname (that
        # would re-resolve unvalidated and reopen the rebinding hole). The host
        # produced no usable address, which is a name-resolution failure, so we
        # surface it as socket.gaierror rather than a bare OSError. gaierror is
        # an OSError subclass, so urllib still wraps it as URLError and the
        # fail-closed contract is unchanged; but callers that legitimately
        # expect a DNS failure -- e.g. obfuscated/decimal-IP hosts that some
        # platforms (Windows) refuse to resolve -- then see the expected
        # gaierror reason instead of an opaque OSError.
        raise socket.gaierror(
            f"pathsec.urlopen: no validated address for host {host!r}; "
            "refusing to connect by unvalidated hostname"
        )
    last_err = None
    for res in addrinfo:
        ip = res[4][0]
        try:
            return socket.create_connection((ip, port), timeout, source_address)
        except OSError as e:
            last_err = e
    raise last_err


class _SafeHTTPConnection(http.client.HTTPConnection):
    """HTTPConnection that resolves, SSRF-validates and pins the address at connect()."""

    def connect(self):
        self.sock = _pinned_connection(
            self.host, self.port, self.timeout, self.source_address
        )


class _SafeHTTPSConnection(http.client.HTTPSConnection):
    """HTTPS variant of :class:`_SafeHTTPConnection`.

    Connects to a validated, pinned IP but keeps SNI / certificate verification
    against the original hostname.
    """

    def connect(self):
        sock = _pinned_connection(
            self.host, self.port, self.timeout, self.source_address
        )
        self.sock = self._context.wrap_socket(sock, server_hostname=self.host)


class _SafeHTTPHandler(urllib.request.HTTPHandler):
    def http_open(self, req):
        return self.do_open(_SafeHTTPConnection, req)


class _SafeHTTPSHandler(urllib.request.HTTPSHandler):
    def https_open(self, req):
        kwargs = {}
        if getattr(self, "_context", None) is not None:
            kwargs["context"] = self._context
        if getattr(self, "_check_hostname", None) is not None:
            kwargs["check_hostname"] = self._check_hostname
        return self.do_open(_SafeHTTPSConnection, req, **kwargs)


def _proxied_fetch_allowed():
    """True only if the operator has explicitly opted into trusting the proxy."""
    if ALLOW_PROXIED_FETCH:
        return True
    return os.environ.get("NLTK_ALLOW_PROXIED_URLOPEN", "").strip().lower() in (
        "1",
        "true",
        "yes",
        "on",
    )


def _reject_unpinnable_proxied_fetch(url_str):
    """Fail closed on a proxied fetch NLTK cannot SSRF-validate (CWE-918).

    A configured proxy performs the egress, so NLTK can neither pin the
    validated IP nor observe the address the proxy ultimately reaches. Rather
    than reintroduce the DNS-rebinding / internal-routing SSRF that the pinned
    direct path closes, refuse the request unless the operator has asserted the
    proxy is trusted (``ALLOW_PROXIED_FETCH`` / ``NLTK_ALLOW_PROXIED_URLOPEN``).
    """
    if _proxied_fetch_allowed():
        return
    msg = (
        f"Security Violation [pathsec.urlopen]: refusing a proxied fetch of "
        f"{url_str!r}. A configured proxy performs the egress, so NLTK cannot "
        "pin the validated IP and SSRF protection cannot be enforced (CWE-918). "
        "If and only if the proxy is trusted to be SSRF-safe, opt in via "
        "NLTK_ALLOW_PROXIED_URLOPEN=1 or nltk.pathsec.ALLOW_PROXIED_FETCH=True."
    )
    if ENFORCE:
        raise PermissionError(msg)
    warnings.warn(msg, RuntimeWarning, stacklevel=3)


def urlopen(url, *args, **kwargs):
    """
    Secure wrapper for urllib.request.urlopen with redirect validation.
    Inherits NLTK proxy settings, but intentionally ignores other custom
    global handlers to strictly enforce the security sandbox.
    """
    url_str = url.full_url if hasattr(url, "full_url") else str(url)
    validate_network_url(url_str, context="pathsec.urlopen")

    # Start with our security-enforcing redirect handler
    handlers = [_ValidatingRedirectHandler()]

    # Safely inherit proxy settings without reusing handler instances
    # (Reusing instances overwrites their .parent, breaking the global opener)
    proxied = False
    has_proxy_handler = False
    if urllib.request._opener is not None:
        for handler in urllib.request._opener.handlers:
            if isinstance(handler, urllib.request.ProxyHandler):
                has_proxy_handler = True
                # Copy the dictionary to prevent shared mutable state
                isolated_proxies = dict(handler.proxies) if handler.proxies else {}
                if isolated_proxies:
                    proxied = True
                handlers.append(urllib.request.ProxyHandler(isolated_proxies))
            elif isinstance(handler, urllib.request.ProxyBasicAuthHandler):
                handlers.append(urllib.request.ProxyBasicAuthHandler(handler.passwd))
            elif isinstance(handler, urllib.request.ProxyDigestAuthHandler):
                handlers.append(urllib.request.ProxyDigestAuthHandler(handler.passwd))

    # If the caller configured no ProxyHandler at all, environment proxies still
    # apply: build_opener() would install a default ProxyHandler from
    # getproxies(). Treat that as proxied too, because the proxy -- not NLTK --
    # is then the egress that resolves names and performs the CONNECT tunnel; the
    # connect-time pinning handlers cannot tunnel and would break proxied HTTPS.
    if not proxied and not has_proxy_handler and urllib.request.getproxies():
        proxied = True

    if not proxied:
        # No proxy in effect: NLTK makes the connection itself, so pin the
        # validated IP (a rebinding hostname cannot be re-resolved to an internal
        # address at connect time). Add an explicit empty ProxyHandler so
        # build_opener() does not silently re-enable environment proxies, which
        # the pinning handlers cannot tunnel through.
        if not has_proxy_handler:
            handlers.append(urllib.request.ProxyHandler({}))
        handlers.append(_SafeHTTPHandler())
        handlers.append(_SafeHTTPSHandler())
    else:
        # Proxied: the proxy resolves the name and performs the egress, so the
        # earlier validate_network_url() only reflects NLTK's *local* view of the
        # host, not what the proxy actually reaches. A rebinding name or a proxy
        # with an internal network view re-opens SSRF (CWE-918, GHSA-6ww7). We
        # cannot pin through a proxy, so fail closed unless the operator has
        # asserted the proxy is trusted.
        _reject_unpinnable_proxied_fetch(url_str)

    opener = urllib.request.build_opener(*handlers)
    return opener.open(url, *args, **kwargs)


def _is_readonly_mode(mode):
    """True for a plain read mode with no create/write/append/update intent."""
    return set(mode) <= set("rbt") and "r" in mode


def _fd_realpath(fd):
    """The kernel's real path for an open descriptor, or ``None`` if unavailable.

    This reflects the inode actually opened, so validating it is race-free: no
    path re-resolution (which an attacker could win) is involved.
    """
    if sys.platform == "darwin":
        try:
            import fcntl

            f_getpath = 50  # <sys/fcntl.h> F_GETPATH
            buf = fcntl.fcntl(fd, f_getpath, b"\x00" * 1024)
            return os.fsdecode(buf.split(b"\x00", 1)[0])
        except (OSError, ValueError):
            return None
    try:  # Linux and other /proc systems
        return os.readlink(f"/proc/self/fd/{fd}")
    except OSError:
        return None


def _os_open_flags(mode):
    """Translate a :func:`builtins.open` *mode* string into ``os.open()`` flags.

    Mirrors the read/write/create/append/truncate intent of *mode* so the
    hardened opener can add ``O_NOFOLLOW`` on top of exactly what the caller
    asked for. POSIX only.
    """
    chars = set(mode)
    if "x" in chars:
        creat = os.O_CREAT | os.O_EXCL
    elif "w" in chars:
        creat = os.O_CREAT | os.O_TRUNC
    elif "a" in chars:
        creat = os.O_CREAT | os.O_APPEND
    else:  # "r" -- open existing, no create
        creat = 0
    if "+" in chars:
        access = os.O_RDWR
    elif chars & {"w", "a", "x"}:
        access = os.O_WRONLY
    else:
        access = os.O_RDONLY
    return access | creat


def _hardened_open(raw_path, mode, context, required_root, **kwargs):
    """Open ``raw_path`` for read *or* write, closing the symlink-swap TOCTOU and
    the hardlink escape that a path-only ``validate_path`` cannot.

    1. Open ``raw_path`` with ``O_NOFOLLOW`` so its final component is never
       followed as a symlink. Unlike opening a re-resolved path, this is atomic:
       the kernel either opens the real inode at that path or fails, so a symlink
       swapped in after the caller's ``validate_path`` cannot redirect the open
       (TOCTOU). On the write side this stops an attacker who plants/swaps
       ``<root>/pkg`` -> ``/etc/...`` between the check and a ``retrieve()`` /
       model write from redirecting the write outside the root -- the symmetric
       counterpart of the read-side content leak (CWE-59, GHSA-f794 class).
       Corpora/data files are never symlinks, so refusing a final-component
       symlink outright is fail-closed.
    2. ``fstat`` the descriptor and refuse ``st_nlink > 1``: a hardlink is another
       name for the same inode with no symlink to resolve, so an in-root hardlink
       to an outside inode would otherwise be read or written through.
    3. Re-validate the *opened descriptor's* real path (from the kernel, not by
       re-resolving the string) so a swapped intermediate directory symlink that
       redirected the open to an outside inode is still caught.

    A newly created file is given ``0600`` permissions so a write into a shared
    temp dir is not left group-/world-readable (CWE-377/378). POSIX only;
    callers fall back to :func:`builtins.open` elsewhere.
    """
    import errno

    flags = (
        _os_open_flags(mode)
        | getattr(os, "O_NOFOLLOW", 0)
        | getattr(os, "O_CLOEXEC", 0)
    )
    try:
        # 0o600 is only consulted when O_CREAT is in flags (write/append/x modes).
        fd = os.open(raw_path, flags, 0o600)
    except OSError as e:
        if e.errno in (errno.ELOOP, errno.EMLINK):
            raise PermissionError(
                f"Security Violation [pathsec.open]: refusing to follow a symlink "
                f"at open time for {raw_path!r} (TOCTOU guard, CWE-59)"
            ) from e
        raise
    try:
        st = os.fstat(fd)
        if st.st_nlink > 1:
            raise PermissionError(
                f"Security Violation [pathsec.open]: refusing multiply-linked file "
                f"{raw_path!r} (st_nlink={st.st_nlink}); a hardlink can point at an "
                "outside-root inode (CWE-59)"
            )
        # Validate what was actually opened (race-free: the fd is pinned). Falls
        # back to the resolved string only if the kernel path is unavailable.
        actual = _fd_realpath(fd)
        validate_path(
            actual if actual is not None else os.path.realpath(raw_path),
            context=context,
            required_root=required_root,
        )
    except BaseException:
        os.close(fd)
        raise
    try:
        return os.fdopen(fd, mode, **kwargs)
    except BaseException:
        # os.open() accepts some mode strings that os.fdopen() rejects (e.g. an
        # empty or malformed mode); without this the descriptor would leak and
        # repeated calls could exhaust the fd table (DoS).
        os.close(fd)
        raise


# Back-compat alias: the opener now handles write modes too, but external
# callers/tests may still import the original read-only name.
_hardened_read_open = _hardened_open


def open(file, mode="r", *, context="pathsec.open", required_root=None, **kwargs):
    """Secure wrapper for builtins.open."""
    # 1. Allow file descriptors (integers) to pass through, matching original logic
    if isinstance(file, int):
        validate_path(file, context=context, required_root=required_root)
        return builtins.open(file, mode=mode, **kwargs)

    # 2a. NLTK PathPointer objects (e.g. FileSystemPathPointer, which is a str
    # subclass) expose a string ``.path``; normalise to that exact string so the
    # strict primitive check below accepts it, matching validate_path's handling.
    if type(file) not in (str, bytes) and isinstance(
        getattr(file, "path", None), (str, bytes)
    ):
        file = file.path

    # 2. Force extraction of the real path from PathLike objects
    try:
        raw_path = os.fspath(file)
    except TypeError:
        raise TypeError("Path must be a string, bytes, or os.PathLike") from None

    # 3. Strict primitive enforcement against type manipulation
    if type(raw_path) not in (str, bytes):
        raise TypeError(
            f"Strict security policy: Path must resolve to exact str or bytes, not '{type(raw_path).__name__}'"
        )

    if type(raw_path) is bytes:
        raw_path = os.fsdecode(raw_path)

    # 4. Execution Substitution: validate and open the pure primitive, discarding the original object
    validate_path(raw_path, context=context, required_root=required_root)
    # 5. Under enforcement on POSIX, additionally close the validate-then-open
    # symlink TOCTOU and the hardlink escape that a path-based check cannot see.
    # This now covers read *and* write/append/update modes: the write side is the
    # symmetric counterpart of the read-side leak -- a symlink swapped in at the
    # destination after validate_path would otherwise let a retrieve()/model write
    # land outside the root (CWE-59). Non-POSIX keeps the plain open.
    if ENFORCE and os.name == "posix":
        # The hardened path owns this open entirely: it raises PermissionError /
        # ValueError on a security violation and the usual OSError (e.g.
        # FileNotFoundError) on a genuine open failure. We must NOT fall back to
        # builtins.open on OSError -- retrying the raw path would follow a symlink
        # the hardened open deliberately refused, reopening the TOCTOU leak.
        return _hardened_open(raw_path, mode, context, required_root, **kwargs)
    return builtins.open(raw_path, mode=mode, **kwargs)


class ZipFile(zipfile.ZipFile):
    """Secure wrapper for zipfile.ZipFile."""

    def __init__(self, file, *args, **kwargs):
        # zipfile.ZipFile also accepts file-like objects (e.g., io.BytesIO).
        # We only strictly normalize and validate path-like objects.
        if isinstance(file, (str, bytes, os.PathLike)):
            try:
                raw_path = os.fspath(file)
            except TypeError:
                raise TypeError(
                    "Path must be a string, bytes, or os.PathLike"
                ) from None

            if type(raw_path) not in (str, bytes):
                raise TypeError(
                    f"Strict security policy: Path must resolve to exact str or bytes, not '{type(raw_path).__name__}'"
                )

            if type(raw_path) is bytes:
                raw_path = os.fsdecode(raw_path)

            validate_path(raw_path, context="pathsec.ZipFile")
            file_to_open = raw_path
        else:
            file_to_open = file

        super().__init__(file_to_open, *args, **kwargs)

    def extract(self, member, path=None, pwd=None):
        validate_zip_archive(self, path or os.getcwd(), specific_member=member)
        return super().extract(member, path, pwd)

    def extractall(self, path=None, members=None, pwd=None):
        validate_zip_archive(self, path or os.getcwd())
        super().extractall(path, members, pwd)


__all__ = [
    "validate_path",
    "validate_network_url",
    "validate_zip_archive",
    "open",
    "urlopen",
    "ZipFile",
    "ENFORCE",
]