""" Dependency-free network-address policy shared by the Sphinx adapter and service. One module decides which feedback endpoints a site may publish and which browser origins a service may allow, so the two sides cannot drift apart. Notes ----- **User notes.** Three policies, from narrow to wide: ``"strict"`` (default) HTTPS on port 443, HTTP or HTTPS (any port) on a loopback host, and a same-origin path such as ``/v1/feedback``. Right for public sites. ``"private-network"`` Adds HTTPS on any port and plain HTTP to private-network hosts: RFC 1918, CGNAT, link-local and unique-local IP literals, single-label names (``feedback``), and names under ``.internal``, ``.local``, ``.localhost``, ``.home.arpa`` or an operator-supplied suffix (``.corp.example``). Right for intranet, air-gapped and on-premises documentation. ``"any"`` Adds plain HTTP to any host. Feedback text then crosses networks unencrypted; use only where the transport is protected some other way. A browser blocks an ``http://`` endpoint from an ``https://`` page (mixed content), whatever the policy says; serve both over the same scheme. **Developer notes.** Private ranges are listed explicitly rather than taken from :attr:`ipaddress.IPv4Address.is_private`, whose answer has changed between Python releases; the decision must be identical on every supported interpreter. Name matching is purely lexical: no DNS lookup is made, so the result never depends on the machine that runs the build. """ from __future__ import annotations import ipaddress import re from typing import Any, Iterable from urllib.parse import urlsplit #: Endpoint/origin policies, narrowest first. NETWORK_POLICIES = ("strict", "private-network", "any") #: Name suffixes reserved for local or private use (RFC 6761 ``.localhost``, #: RFC 6762 ``.local``, RFC 8375 ``.home.arpa``, ICANN-reserved ``.internal``). RESERVED_PRIVATE_SUFFIXES = (".localhost", ".local", ".internal", ".home.arpa") _PRIVATE_NETWORKS = tuple( ipaddress.ip_network(cidr) for cidr in ( "10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16", "100.64.0.0/10", # carrier-grade NAT / overlay VPNs "169.254.0.0/16", # link-local "fc00::/7", # unique-local "fe80::/10", # link-local ) ) _LOOPBACK_NETWORKS = tuple( ipaddress.ip_network(cidr) for cidr in ("127.0.0.0/8", "::1/128") ) _LABEL_RE = re.compile(r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\Z") _MAX_URL_CHARS = 2048 _MAX_SUFFIXES = 32 class NetworkPolicyError(ValueError): """Raised for an address the selected policy does not permit.""" def normalize_policy(value: Any, *, name: str) -> str: """ Return a canonical policy name. Parameters ---------- value : str or None ``None`` or ``""`` selects ``"strict"``. name : str Setting name used in error messages. Returns ------- str One of :data:`NETWORK_POLICIES`. Raises ------ NetworkPolicyError If *value* is not a string naming a known policy. """ if value in (None, ""): return "strict" if not isinstance(value, str): raise NetworkPolicyError(f"{name} must be a string") text = value.strip().lower() if text not in NETWORK_POLICIES: raise NetworkPolicyError(f"{name} must be one of {list(NETWORK_POLICIES)}") return text def normalize_private_suffixes(value: Any, *, name: str) -> tuple[str, ...]: """ Return operator-supplied private name suffixes in canonical form. Parameters ---------- value : str, list of str, tuple of str or None Comma-separated string or sequence, for example ``".corp.example"``. A missing leading dot is added. name : str Setting name used in error messages. Returns ------- tuple of str Lowercase suffixes, each starting with ``.``, without duplicates. Raises ------ NetworkPolicyError If an entry is not a dot-separated sequence of DNS labels, or there are more than 32 entries. """ if value in (None, "", (), []): return () if isinstance(value, str): items: Iterable[Any] = value.split(",") elif isinstance(value, (list, tuple)): items = value else: raise NetworkPolicyError(f"{name} must be a string or a list of strings") result: list[str] = [] for item in items: if not isinstance(item, str): raise NetworkPolicyError(f"{name} entries must be strings") text = item.strip().lower() if not text: continue if not text.startswith("."): text = "." + text labels = text[1:].split(".") if not all(_LABEL_RE.fullmatch(label) for label in labels): raise NetworkPolicyError(f"{name} contains an invalid suffix: {item!r}") if text not in result: result.append(text) if len(result) > _MAX_SUFFIXES: raise NetworkPolicyError(f"{name} contains more than {_MAX_SUFFIXES} suffixes") return tuple(result) def _ip(host: str): try: return ipaddress.ip_address(host.strip("[]")) except ValueError: return None def is_loopback_host(host: str) -> bool: """Return whether *host* is ``localhost``, ``*.localhost`` or a loopback IP.""" text = (host or "").lower().rstrip(".") address = _ip(text) if address is not None: return any(address in network for network in _LOOPBACK_NETWORKS) return text == "localhost" or text.endswith(".localhost") def is_private_host(host: str, extra_suffixes: Iterable[str] = ()) -> bool: """ Return whether *host* names a loopback or private-network destination. Parameters ---------- host : str Host as returned by :func:`urllib.parse.urlsplit` ``hostname``. extra_suffixes : iterable of str Canonical suffixes from :func:`normalize_private_suffixes`. Returns ------- bool ``True`` for loopback and private IP literals, single-label names, and names under a reserved or extra suffix. No DNS lookup is made. """ text = (host or "").lower().rstrip(".") if not text: return False if is_loopback_host(text): return True address = _ip(text) if address is not None: return any(address in network for network in _PRIVATE_NETWORKS) if "." not in text: return True return text.endswith(tuple(RESERVED_PRIVATE_SUFFIXES) + tuple(extra_suffixes)) def _check_text(value: Any, *, name: str) -> str: if not isinstance(value, str): raise NetworkPolicyError(f"{name} must be a string") text = value.strip() if len(text) > _MAX_URL_CHARS or any( ord(ch) < 33 or ord(ch) == 127 # ruff: ignore[magic-value-comparison] for ch in text ): raise NetworkPolicyError(f"{name} contains whitespace or control characters") return text def _same_origin_path(text: str, *, name: str) -> str: if text.startswith("//") or any(marker in text for marker in ("\\", "?", "#")): raise NetworkPolicyError( f"{name} same-origin path must be a plain absolute path such as " "'/v1/feedback'" ) if any(part in {".", ".."} for part in text.split("/")[1:]): raise NetworkPolicyError(f"{name} same-origin path must not contain . or ..") return text.rstrip("/") or "/" def _scheme_allowed( scheme: str, host: str, port: int | None, *, policy: str, suffixes: Iterable[str] ) -> bool: if is_loopback_host(host): return scheme in {"http", "https"} if scheme == "https": return port in (None, 443) or policy != "strict" if scheme == "http": if policy == "any": return True return policy == "private-network" and is_private_host(host, suffixes) return False def validate_endpoint_url( value: Any, *, policy: str = "strict", private_suffixes: Iterable[str] = (), name: str = "feedback_endpoint", ) -> str: """ Validate a browser-visible feedback endpoint under *policy*. Parameters ---------- value : str ``""`` (no endpoint), a same-origin path starting with one ``/``, or an absolute ``http``/``https`` URL. policy : {"strict", "private-network", "any"}, default "strict" See the module notes. private_suffixes : iterable of str, default () Extra private name suffixes for ``"private-network"``. name : str, default "feedback_endpoint" Setting name used in error messages. Returns ------- str The endpoint without a trailing slash, or ``""``. Raises ------ NetworkPolicyError For credentials, a query or fragment, an unknown scheme, a missing host, an invalid port, or a scheme/host/port the policy does not allow. Examples -------- >>> validate_endpoint_url("/v1/feedback") '/v1/feedback' >>> validate_endpoint_url( ... "http://feedback.internal:8080/v1/feedback", policy="private-network" ... ) 'http://feedback.internal:8080/v1/feedback' """ text = _check_text(value, name=name) if not text: return "" if text.startswith("/"): return _same_origin_path(text, name=name) try: parsed = urlsplit(text) port = parsed.port except ValueError as exc: raise NetworkPolicyError(f"{name} is not a valid URL") from exc host = (parsed.hostname or "").lower() if parsed.scheme not in {"http", "https"}: raise NetworkPolicyError(f"{name} must be an http(s) URL or a same-origin path") if parsed.username or parsed.password or parsed.fragment or parsed.query: raise NetworkPolicyError( f"{name} must not contain credentials, a query, or a fragment" ) if not host: raise NetworkPolicyError(f"{name} must include a host") if not _scheme_allowed( parsed.scheme, host, port, policy=policy, suffixes=private_suffixes ): raise NetworkPolicyError(_refusal(name, parsed.scheme, policy)) return text.rstrip("/") def _refusal(name: str, scheme: str, policy: str) -> str: if policy == "strict": return ( f"{name} must use HTTPS on the standard port (localhost HTTP is " "allowed); set the policy to 'private-network' for intranet hosts" ) if policy == "private-network": return ( f"{name} uses {scheme} to a host that is not on a private network; " "use HTTPS, add the host's suffix to the private suffixes, or set " "the policy to 'any'" ) return f"{name} is not allowed" def normalize_origin( value: Any, *, policy: str = "strict", private_suffixes: Iterable[str] = (), name: str = "feedback allowed origin", ) -> str: """ Return a canonical browser origin (``scheme://host[:port]``). Parameters ---------- value : str An origin; a trailing ``/`` is accepted, any other path is not. policy, private_suffixes, name As for :func:`validate_endpoint_url`. Returns ------- str Lowercase host, default ports (443/80) dropped, IPv6 bracketed. Raises ------ NetworkPolicyError For anything that is not a bare origin, or that *policy* forbids. """ text = _check_text(value, name=name) if not text: raise NetworkPolicyError(f"{name} is invalid") try: parsed = urlsplit(text) port = parsed.port except ValueError as exc: raise NetworkPolicyError(f"{name} is invalid") from exc host = (parsed.hostname or "").lower() if ( parsed.scheme not in {"http", "https"} or not host or parsed.username or parsed.password or parsed.path not in {"", "/"} or parsed.query or parsed.fragment ): raise NetworkPolicyError(f"{name} must contain only scheme and authority") if not _scheme_allowed( parsed.scheme, host, port, policy=policy, suffixes=private_suffixes ): raise NetworkPolicyError(_refusal(name, parsed.scheme, policy)) authority = f"[{host}]" if ":" in host else host default_port = 443 if parsed.scheme == "https" else 80 if port is not None and port != default_port: authority += f":{port}" return f"{parsed.scheme}://{authority}" __all__ = [ "NETWORK_POLICIES", "RESERVED_PRIVATE_SUFFIXES", "NetworkPolicyError", "is_loopback_host", "is_private_host", "normalize_origin", "normalize_policy", "normalize_private_suffixes", "validate_endpoint_url", ]