Spaces:

User1342
/

distinct

Sleeping

distinct / distinct_protocol /netpolicy.py

initial commit

2aa8b3a 2 months ago

15.7 kB

"""Network binding and connection policy: loopback by default, wider on request.
**This is a default, not a prohibition.** distinct is a multi-machine system —
agents on contributors' desktops polling a server they do not host — and the
code must not forbid the configuration it exists to serve.  What it does
instead is refuse to reach that configuration *by accident*:
* nothing binds beyond loopback unless an operator asked for it, in a named
  place, on purpose;
* the ask is loud — the exact exposure is printed at start-up, every time;
* accidental widening (a literal ``0.0.0.0`` typed into a source file, a
  ``share=True`` left in a debug session) is caught by tests, while a
  legitimate production configuration passes them, because a production
  configuration supplies a value at runtime rather than hard-coding one.
Two things it does still hold firm on, and both are stated so they can be
argued with rather than discovered:
1. **A non-loopback server URL must be HTTPS.** The agent signs requests with a
   shared secret and receives prompts it will execute. Over plain HTTP across a
   network, anything on the path can read and rewrite both.
2. **Gradio's ``share=True`` tunnel stays off.** Binding a LAN or public
   interface is the operator publishing their own service on their own network.
   A share tunnel publishes it to the public internet through a third-party
   relay, which is a different decision with a different blast radius. If that
   is wanted, it should be its own explicit switch rather than arriving as a
   side effect of this one.
Configuration
-------------
===============================  ============================================
``DISTINCT_BIND_HOST``           Server listen address. Default ``127.0.0.1``.
                                 ``0.0.0.0`` binds every interface.
``DISTINCT_ALLOW_REMOTE_SERVER`` Set to ``1``/``true``/``yes`` to let an agent
                                 connect to a non-loopback server. The agent's
                                 ``--allow-remote-server`` flag does the same.
``PORT``                         Server port. Default ``7860``.
===============================  ============================================
"""

from __future__ import annotations

import ipaddress
import os
import sys
from collections.abc import Mapping
from urllib.parse import urlsplit

#: What anything binds to unless told otherwise.
DEFAULT_BIND_HOST = "127.0.0.1"

#: Kept as a name because "is this the loopback default?" is a question the
#: rest of the code asks, distinct from "what are we binding to?".
LOOPBACK_BIND_HOST = "127.0.0.1"

#: Binds every interface. Named so it can be recognised, not so it is refused.
WILDCARD_BIND_HOSTS = frozenset({"0.0.0.0", "::", "*"})

DEFAULT_SERVER_URL = "http://127.0.0.1:7860"
DEFAULT_SERVER_PORT = 7860

#: Treated as loopback without a DNS lookup.  Resolution is deliberately not
#: attempted: a name that resolves to 127.0.0.1 today can resolve elsewhere
#: tomorrow, and DNS is not a trust boundary.
LOOPBACK_HOSTNAMES = frozenset({"localhost", "localhost.", "ip6-localhost"})

BIND_HOST_ENV_VAR = "DISTINCT_BIND_HOST"
ALLOW_REMOTE_SERVER_ENV_VAR = "DISTINCT_ALLOW_REMOTE_SERVER"
#: A second, separate opt-in for a second, separate risk.
#:
#: Reaching a machine that is not this one and reaching it without TLS are two
#: different decisions, and collapsing them into one flag means an operator who
#: wanted the first silently gets the second. They are asked for separately.
#:
#: The case this exists for is a link that is already encrypted underneath:
#: a WireGuard or Tailscale address carries the traffic inside a tunnel, so
#: TLS on top of it defends against nothing the tunnel does not already cover.
#: It is not a general permission to send prompts over open wifi, and the
#: warning printed when it is used says so.
ALLOW_PLAINTEXT_SERVER_ENV_VAR = "DISTINCT_ALLOW_PLAINTEXT_SERVER"
_TRUTHY = frozenset({"1", "true", "yes", "on"})


class NetworkPolicyError(ValueError):
    """A bind address or server URL is malformed, unsafe, or not permitted."""


def _environ(environ: Mapping[str, str] | None) -> Mapping[str, str]:
    return os.environ if environ is None else environ


def remote_server_allowed(environ: Mapping[str, str] | None = None) -> bool:
    """True when an agent has been configured to use a non-loopback server."""

    return _environ(environ).get(ALLOW_REMOTE_SERVER_ENV_VAR, "").strip().casefold() in _TRUTHY


def plaintext_server_allowed(environ: Mapping[str, str] | None = None) -> bool:
    """True when an operator has accepted a remote server without TLS."""

    return _environ(environ).get(ALLOW_PLAINTEXT_SERVER_ENV_VAR, "").strip().casefold() in _TRUTHY


def is_loopback_host(value: object) -> bool:
    """Return True only for an address that cannot be reached off-machine.
    Literal IPs are decided by :mod:`ipaddress`, so ``127.0.0.2`` and ``::1``
    both pass and ``0.0.0.0`` (unspecified — every interface) does not.  Names
    are matched against an exact allowlist, so ``127.0.0.1.example.com`` and
    ``localhost.attacker.test`` are correctly not treated as local.
    """

    if not isinstance(value, str) or not value.strip():
        return False
    host = value.strip()
    if host.startswith("[") and host.endswith("]"):
        host = host[1:-1]
    if "%" in host:  # IPv6 zone identifier, e.g. fe80::1%eth0
        host = host.split("%", 1)[0]
    try:
        address = ipaddress.ip_address(host)
    except ValueError:
        return host.casefold() in LOOPBACK_HOSTNAMES
    return address.is_loopback


def local_addresses(port: int, bind_host: str = DEFAULT_BIND_HOST) -> list[str]:
    """Every URL a listener on this host can actually be reached on.
    Lives here rather than beside any one caller because it answers a
    network-policy question, and because two callers need it: a server telling
    a person which address to point an agent at, and an agent telling a person
    which address to open. Both were previously reduced to a placeholder that
    asked the reader to go and find something the machine already knew.
    Bound to loopback there is exactly one answer and it is the only true one,
    so no amount of interface enumeration would help. Bound wider, the useful
    answers are this host's addresses on its own networks, found by asking the
    routing table which local address it would use to reach the outside world.
    That is a question with an answer even when offline and it sends no
    packet: a UDP socket is connectionless, so ``connect`` only fixes the
    local end.
    Loopback and link-local addresses are filtered out of the wider case
    because typing one into another device does not work, and an address that
    does not work is worse than none.
    """

    import socket

    if is_loopback_host(bind_host):
        return [f"http://{bind_host}:{port}"]

    found: list[str] = []
    for probe in ("8.8.8.8", "192.168.1.1"):
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        try:
            sock.connect((probe, 80))
            address = sock.getsockname()[0]
        except OSError:
            continue
        finally:
            sock.close()
        if address and not address.startswith(("127.", "169.254.")):
            found.append(address)
    try:
        for info in socket.getaddrinfo(socket.gethostname(), None, socket.AF_INET):
            address = info[4][0]
            if address and not address.startswith(("127.", "169.254.")):
                found.append(address)
    except OSError:
        pass
    ordered: list[str] = []
    for address in found:
        url = f"http://{address}:{port}"
        if url not in ordered:
            ordered.append(url)
    # Nothing usable found. Saying so beats inventing an address.
    return ordered


def describe_exposure(host: str) -> str:
    """Say in plain words what binding to ``host`` makes reachable."""

    if is_loopback_host(host):
        return "this machine only"
    if host in WILDCARD_BIND_HOSTS:
        return "every network interface on this machine — LAN, VPN, and any public address it has"
    try:
        address = ipaddress.ip_address(host)
    except ValueError:
        return f"whatever {host!r} resolves to"
    # is_global, not "not is_private": Python also marks reserved ranges such
    # as TEST-NET as non-global, and calling those "the public internet" would
    # be wrong.
    if address.is_global:
        return f"the public internet via {host}"
    return f"the local network via {host}"


def announce_bind(host: str, port: int, *, stream: object = None) -> str:
    """Print, and return, the one line an operator must be able to trust."""

    exposure = describe_exposure(host)
    if is_loopback_host(host):
        message = f"distinct server listening on http://{host}:{port} — reachable from {exposure}."
    else:
        message = (
            f"distinct server listening on http://{host}:{port} — REACHABLE FROM {exposure.upper()}. "
            f"This was requested explicitly ({BIND_HOST_ENV_VAR}={host}). Anyone who can reach this "
            "address can submit prompts, see your agents' hardware and energy telemetry, and spend "
            "your workers' compute. Serve it behind TLS and an authenticating proxy."
        )
    print(message, file=stream or (sys.stdout if is_loopback_host(host) else sys.stderr), flush=True)
    return message


def validate_bind_host(value: object) -> str:
    """Accept any syntactically valid bind address, including wildcards."""

    if not isinstance(value, str) or not value.strip():
        raise NetworkPolicyError("bind address must be a non-empty host")
    host = value.strip()
    if host in WILDCARD_BIND_HOSTS or is_loopback_host(host):
        return host
    try:
        ipaddress.ip_address(host.strip("[]").split("%", 1)[0])
    except ValueError as exc:
        # A hostname is fine for a client URL but meaningless for a listener:
        # a socket binds an address, not a name.
        raise NetworkPolicyError(
            f"bind address {host!r} is not an IP address or a recognised wildcard"
        ) from exc
    return host


def resolve_bind_host(
    requested: object = None,
    *,
    environ: Mapping[str, str] | None = None,
) -> str:
    """Return the host a listener should bind to.
    Precedence: an explicit argument, then :data:`BIND_HOST_ENV_VAR`, then
    :data:`DEFAULT_BIND_HOST`.  The default is loopback, so a deployment is
    never wider than someone asked for.
    """

    if requested is None:
        requested = _environ(environ).get(BIND_HOST_ENV_VAR) or None
    if requested is None:
        return DEFAULT_BIND_HOST
    return validate_bind_host(requested)


def resolve_bind_port(
    requested: object = None,
    *,
    environ: Mapping[str, str] | None = None,
) -> int:
    """Return a validated TCP port, defaulting to :data:`DEFAULT_SERVER_PORT`."""

    if requested is None:
        raw = _environ(environ).get("PORT")
        if raw is None or not str(raw).strip():
            return DEFAULT_SERVER_PORT
        requested = raw
    try:
        port = int(str(requested).strip())
    except (TypeError, ValueError) as exc:
        raise NetworkPolicyError("port must be an integer") from exc
    if not 1 <= port <= 65_535:
        raise NetworkPolicyError("port must be between 1 and 65535")
    return port


def launch_kwargs(
    *,
    server_port: object = None,
    server_name: object = None,
    environ: Mapping[str, str] | None = None,
) -> dict[str, object]:
    """Build the network-relevant kwargs for ``gradio.Blocks.launch``.
    ``share=False`` is returned explicitly rather than left to Gradio's
    default.  Binding a wider interface is the operator publishing on their own
    network; a share tunnel publishes through a third-party relay to the public
    internet, and that is a separate decision — see the module docstring.
    """

    return {
        "server_name": resolve_bind_host(server_name, environ=environ),
        "server_port": resolve_bind_port(server_port, environ=environ),
        "share": False,
    }


def validate_server_url(
    value: object,
    *,
    what: str = "server URL",
    allow_remote: bool | None = None,
    allow_plaintext: bool | None = None,
    environ: Mapping[str, str] | None = None,
) -> str:
    """Validate an outbound URL and apply the loopback default.
    A loopback URL is always accepted, over plain HTTP, because TLS there
    protects against nothing a local attacker could not already do.  A
    non-loopback URL requires (a) an explicit opt-in and (b) HTTPS.
    """

    if not isinstance(value, str) or not value.strip():
        raise NetworkPolicyError(f"{what} must be non-empty text")
    url = value.strip().rstrip("/")
    parsed = urlsplit(url)
    if parsed.username is not None or parsed.password is not None:
        raise NetworkPolicyError(f"{what} must not contain credentials")
    if parsed.query or parsed.fragment:
        raise NetworkPolicyError(f"{what} must not contain a query or fragment")
    if not parsed.hostname:
        raise NetworkPolicyError(f"{what} must contain a host")
    if parsed.scheme not in {"http", "https"}:
        raise NetworkPolicyError(f"{what} must use http or https")

    if is_loopback_host(parsed.hostname):
        return url

    permitted = remote_server_allowed(environ) if allow_remote is None else bool(allow_remote)
    if not permitted:
        raise NetworkPolicyError(
            f"{what} {url!r} points at {parsed.hostname!r}, which is not on this machine. "
            "Remote servers are supported but must be opted into, so that a typo cannot send "
            f"your prompts and telemetry somewhere unintended. Pass --allow-remote-server, or "
            f"set {ALLOW_REMOTE_SERVER_ENV_VAR}=1."
        )
    plaintext = (
        plaintext_server_allowed(environ) if allow_plaintext is None else bool(allow_plaintext)
    )
    if parsed.scheme != "https" and not plaintext:
        raise NetworkPolicyError(
            f"{what} must use HTTPS when the server is not on this machine: the agent's "
            "signing secret and the prompts it executes would otherwise cross the network "
            "in the clear. If the link is already encrypted underneath, a WireGuard or "
            "Tailscale address for instance, pass --allow-plaintext-server or set "
            f"{ALLOW_PLAINTEXT_SERVER_ENV_VAR}=1 to say so."
        )
    return url


def announce_remote_server(url: str, *, stream: object = None) -> str:
    """Print, and return, what connecting to a remote server grants it."""

    message = (
        f"Agent will poll REMOTE server {url}. Its users can send arbitrary prompts to this "
        "machine, will see your OS, CPU, RAM, model list and energy telemetry, and can queue "
        "work that competes for your inference slot. You opted into this explicitly."
    )
    print(message, file=stream or sys.stderr, flush=True)
    return message


# Retained under the old name so existing call sites keep working.
validate_loopback_url = validate_server_url


__all__ = [
    "ALLOW_REMOTE_SERVER_ENV_VAR",
    "BIND_HOST_ENV_VAR",
    "DEFAULT_BIND_HOST",
    "DEFAULT_SERVER_PORT",
    "DEFAULT_SERVER_URL",
    "LOOPBACK_BIND_HOST",
    "LOOPBACK_HOSTNAMES",
    "WILDCARD_BIND_HOSTS",
    "NetworkPolicyError",
    "announce_bind",
    "announce_remote_server",
    "describe_exposure",
    "is_loopback_host",
    "launch_kwargs",
    "remote_server_allowed",
    "resolve_bind_host",
    "resolve_bind_port",
    "validate_bind_host",
    "validate_loopback_url",
    "validate_server_url",
]

	"""Network binding and connection policy: loopback by default, wider on request.

	This is a default, not a prohibition. distinct is a multi-machine system —
	agents on contributors' desktops polling a server they do not host — and the
	code must not forbid the configuration it exists to serve. What it does
	instead is refuse to reach that configuration by accident:

	* nothing binds beyond loopback unless an operator asked for it, in a named
	place, on purpose;
	* the ask is loud — the exact exposure is printed at start-up, every time;
	* accidental widening (a literal ``0.0.0.0`` typed into a source file, a
	``share=True`` left in a debug session) is caught by tests, while a
	legitimate production configuration passes them, because a production
	configuration supplies a value at runtime rather than hard-coding one.

	Two things it does still hold firm on, and both are stated so they can be
	argued with rather than discovered:

	1. A non-loopback server URL must be HTTPS. The agent signs requests with a
	shared secret and receives prompts it will execute. Over plain HTTP across a
	network, anything on the path can read and rewrite both.
	2. Gradio's ``share=True`` tunnel stays off. Binding a LAN or public
	interface is the operator publishing their own service on their own network.
	A share tunnel publishes it to the public internet through a third-party
	relay, which is a different decision with a different blast radius. If that
	is wanted, it should be its own explicit switch rather than arriving as a
	side effect of this one.

	Configuration
	-------------

	=============================== ============================================
	``DISTINCT_BIND_HOST`` Server listen address. Default ``127.0.0.1``.
	``0.0.0.0`` binds every interface.
	``DISTINCT_ALLOW_REMOTE_SERVER`` Set to ``1``/``true``/``yes`` to let an agent
	connect to a non-loopback server. The agent's
	``--allow-remote-server`` flag does the same.
	``PORT`` Server port. Default ``7860``.
	=============================== ============================================
	"""

	from __future__ import annotations

	import ipaddress
	import os
	import sys
	from collections.abc import Mapping
	from urllib.parse import urlsplit

	#: What anything binds to unless told otherwise.
	DEFAULT_BIND_HOST = "127.0.0.1"

	#: Kept as a name because "is this the loopback default?" is a question the
	#: rest of the code asks, distinct from "what are we binding to?".
	LOOPBACK_BIND_HOST = "127.0.0.1"

	#: Binds every interface. Named so it can be recognised, not so it is refused.
	WILDCARD_BIND_HOSTS = frozenset({"0.0.0.0", "::", "*"})

	DEFAULT_SERVER_URL = "http://127.0.0.1:7860"
	DEFAULT_SERVER_PORT = 7860

	#: Treated as loopback without a DNS lookup. Resolution is deliberately not
	#: attempted: a name that resolves to 127.0.0.1 today can resolve elsewhere
	#: tomorrow, and DNS is not a trust boundary.
	LOOPBACK_HOSTNAMES = frozenset({"localhost", "localhost.", "ip6-localhost"})

	BIND_HOST_ENV_VAR = "DISTINCT_BIND_HOST"
	ALLOW_REMOTE_SERVER_ENV_VAR = "DISTINCT_ALLOW_REMOTE_SERVER"
	#: A second, separate opt-in for a second, separate risk.
	#:
	#: Reaching a machine that is not this one and reaching it without TLS are two
	#: different decisions, and collapsing them into one flag means an operator who
	#: wanted the first silently gets the second. They are asked for separately.
	#:
	#: The case this exists for is a link that is already encrypted underneath:
	#: a WireGuard or Tailscale address carries the traffic inside a tunnel, so
	#: TLS on top of it defends against nothing the tunnel does not already cover.
	#: It is not a general permission to send prompts over open wifi, and the
	#: warning printed when it is used says so.
	ALLOW_PLAINTEXT_SERVER_ENV_VAR = "DISTINCT_ALLOW_PLAINTEXT_SERVER"
	_TRUTHY = frozenset({"1", "true", "yes", "on"})


	class NetworkPolicyError(ValueError):
	"""A bind address or server URL is malformed, unsafe, or not permitted."""


	def _environ(environ: Mapping[str, str] \| None) -> Mapping[str, str]:
	return os.environ if environ is None else environ


	def remote_server_allowed(environ: Mapping[str, str] \| None = None) -> bool:
	"""True when an agent has been configured to use a non-loopback server."""

	return _environ(environ).get(ALLOW_REMOTE_SERVER_ENV_VAR, "").strip().casefold() in _TRUTHY


	def plaintext_server_allowed(environ: Mapping[str, str] \| None = None) -> bool:
	"""True when an operator has accepted a remote server without TLS."""

	return _environ(environ).get(ALLOW_PLAINTEXT_SERVER_ENV_VAR, "").strip().casefold() in _TRUTHY


	def is_loopback_host(value: object) -> bool:
	"""Return True only for an address that cannot be reached off-machine.

	Literal IPs are decided by :mod:`ipaddress`, so ``127.0.0.2`` and ``::1``
	both pass and ``0.0.0.0`` (unspecified — every interface) does not. Names
	are matched against an exact allowlist, so ``127.0.0.1.example.com`` and
	``localhost.attacker.test`` are correctly not treated as local.
	"""

	if not isinstance(value, str) or not value.strip():
	return False
	host = value.strip()
	if host.startswith("[") and host.endswith("]"):
	host = host[1:-1]
	if "%" in host: # IPv6 zone identifier, e.g. fe80::1%eth0
	host = host.split("%", 1)[0]
	try:
	address = ipaddress.ip_address(host)
	except ValueError:
	return host.casefold() in LOOPBACK_HOSTNAMES
	return address.is_loopback


	def local_addresses(port: int, bind_host: str = DEFAULT_BIND_HOST) -> list[str]:
	"""Every URL a listener on this host can actually be reached on.

	Lives here rather than beside any one caller because it answers a
	network-policy question, and because two callers need it: a server telling
	a person which address to point an agent at, and an agent telling a person
	which address to open. Both were previously reduced to a placeholder that
	asked the reader to go and find something the machine already knew.

	Bound to loopback there is exactly one answer and it is the only true one,
	so no amount of interface enumeration would help. Bound wider, the useful
	answers are this host's addresses on its own networks, found by asking the
	routing table which local address it would use to reach the outside world.
	That is a question with an answer even when offline and it sends no
	packet: a UDP socket is connectionless, so ``connect`` only fixes the
	local end.

	Loopback and link-local addresses are filtered out of the wider case
	because typing one into another device does not work, and an address that
	does not work is worse than none.
	"""

	import socket

	if is_loopback_host(bind_host):
	return [f"http://{bind_host}:{port}"]

	found: list[str] = []
	for probe in ("8.8.8.8", "192.168.1.1"):
	sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
	try:
	sock.connect((probe, 80))
	address = sock.getsockname()[0]
	except OSError:
	continue
	finally:
	sock.close()
	if address and not address.startswith(("127.", "169.254.")):
	found.append(address)
	try:
	for info in socket.getaddrinfo(socket.gethostname(), None, socket.AF_INET):
	address = info[4][0]
	if address and not address.startswith(("127.", "169.254.")):
	found.append(address)
	except OSError:
	pass
	ordered: list[str] = []
	for address in found:
	url = f"http://{address}:{port}"
	if url not in ordered:
	ordered.append(url)
	# Nothing usable found. Saying so beats inventing an address.
	return ordered


	def describe_exposure(host: str) -> str:
	"""Say in plain words what binding to ``host`` makes reachable."""

	if is_loopback_host(host):
	return "this machine only"
	if host in WILDCARD_BIND_HOSTS:
	return "every network interface on this machine — LAN, VPN, and any public address it has"
	try:
	address = ipaddress.ip_address(host)
	except ValueError:
	return f"whatever {host!r} resolves to"
	# is_global, not "not is_private": Python also marks reserved ranges such
	# as TEST-NET as non-global, and calling those "the public internet" would
	# be wrong.
	if address.is_global:
	return f"the public internet via {host}"
	return f"the local network via {host}"


	def announce_bind(host: str, port: int, *, stream: object = None) -> str:
	"""Print, and return, the one line an operator must be able to trust."""

	exposure = describe_exposure(host)
	if is_loopback_host(host):
	message = f"distinct server listening on http://{host}:{port} — reachable from {exposure}."
	else:
	message = (
	f"distinct server listening on http://{host}:{port} — REACHABLE FROM {exposure.upper()}. "
	f"This was requested explicitly ({BIND_HOST_ENV_VAR}={host}). Anyone who can reach this "
	"address can submit prompts, see your agents' hardware and energy telemetry, and spend "
	"your workers' compute. Serve it behind TLS and an authenticating proxy."
	)
	print(message, file=stream or (sys.stdout if is_loopback_host(host) else sys.stderr), flush=True)
	return message


	def validate_bind_host(value: object) -> str:
	"""Accept any syntactically valid bind address, including wildcards."""

	if not isinstance(value, str) or not value.strip():
	raise NetworkPolicyError("bind address must be a non-empty host")
	host = value.strip()
	if host in WILDCARD_BIND_HOSTS or is_loopback_host(host):
	return host
	try:
	ipaddress.ip_address(host.strip("[]").split("%", 1)[0])
	except ValueError as exc:
	# A hostname is fine for a client URL but meaningless for a listener:
	# a socket binds an address, not a name.
	raise NetworkPolicyError(
	f"bind address {host!r} is not an IP address or a recognised wildcard"
	) from exc
	return host


	def resolve_bind_host(
	requested: object = None,
	*,
	environ: Mapping[str, str] \| None = None,
	) -> str:
	"""Return the host a listener should bind to.

	Precedence: an explicit argument, then :data:`BIND_HOST_ENV_VAR`, then
	:data:`DEFAULT_BIND_HOST`. The default is loopback, so a deployment is
	never wider than someone asked for.
	"""

	if requested is None:
	requested = _environ(environ).get(BIND_HOST_ENV_VAR) or None
	if requested is None:
	return DEFAULT_BIND_HOST
	return validate_bind_host(requested)


	def resolve_bind_port(
	requested: object = None,
	*,
	environ: Mapping[str, str] \| None = None,
	) -> int:
	"""Return a validated TCP port, defaulting to :data:`DEFAULT_SERVER_PORT`."""

	if requested is None:
	raw = _environ(environ).get("PORT")
	if raw is None or not str(raw).strip():
	return DEFAULT_SERVER_PORT
	requested = raw
	try:
	port = int(str(requested).strip())
	except (TypeError, ValueError) as exc:
	raise NetworkPolicyError("port must be an integer") from exc
	if not 1 <= port <= 65_535:
	raise NetworkPolicyError("port must be between 1 and 65535")
	return port


	def launch_kwargs(
	*,
	server_port: object = None,
	server_name: object = None,
	environ: Mapping[str, str] \| None = None,
	) -> dict[str, object]:
	"""Build the network-relevant kwargs for ``gradio.Blocks.launch``.

	``share=False`` is returned explicitly rather than left to Gradio's
	default. Binding a wider interface is the operator publishing on their own
	network; a share tunnel publishes through a third-party relay to the public
	internet, and that is a separate decision — see the module docstring.
	"""

	return {
	"server_name": resolve_bind_host(server_name, environ=environ),
	"server_port": resolve_bind_port(server_port, environ=environ),
	"share": False,
	}


	def validate_server_url(
	value: object,
	*,
	what: str = "server URL",
	allow_remote: bool \| None = None,
	allow_plaintext: bool \| None = None,
	environ: Mapping[str, str] \| None = None,
	) -> str:
	"""Validate an outbound URL and apply the loopback default.

	A loopback URL is always accepted, over plain HTTP, because TLS there
	protects against nothing a local attacker could not already do. A
	non-loopback URL requires (a) an explicit opt-in and (b) HTTPS.
	"""

	if not isinstance(value, str) or not value.strip():
	raise NetworkPolicyError(f"{what} must be non-empty text")
	url = value.strip().rstrip("/")
	parsed = urlsplit(url)
	if parsed.username is not None or parsed.password is not None:
	raise NetworkPolicyError(f"{what} must not contain credentials")
	if parsed.query or parsed.fragment:
	raise NetworkPolicyError(f"{what} must not contain a query or fragment")
	if not parsed.hostname:
	raise NetworkPolicyError(f"{what} must contain a host")
	if parsed.scheme not in {"http", "https"}:
	raise NetworkPolicyError(f"{what} must use http or https")

	if is_loopback_host(parsed.hostname):
	return url

	permitted = remote_server_allowed(environ) if allow_remote is None else bool(allow_remote)
	if not permitted:
	raise NetworkPolicyError(
	f"{what} {url!r} points at {parsed.hostname!r}, which is not on this machine. "
	"Remote servers are supported but must be opted into, so that a typo cannot send "
	f"your prompts and telemetry somewhere unintended. Pass --allow-remote-server, or "
	f"set {ALLOW_REMOTE_SERVER_ENV_VAR}=1."
	)
	plaintext = (
	plaintext_server_allowed(environ) if allow_plaintext is None else bool(allow_plaintext)
	)
	if parsed.scheme != "https" and not plaintext:
	raise NetworkPolicyError(
	f"{what} must use HTTPS when the server is not on this machine: the agent's "
	"signing secret and the prompts it executes would otherwise cross the network "
	"in the clear. If the link is already encrypted underneath, a WireGuard or "
	"Tailscale address for instance, pass --allow-plaintext-server or set "
	f"{ALLOW_PLAINTEXT_SERVER_ENV_VAR}=1 to say so."
	)
	return url


	def announce_remote_server(url: str, *, stream: object = None) -> str:
	"""Print, and return, what connecting to a remote server grants it."""

	message = (
	f"Agent will poll REMOTE server {url}. Its users can send arbitrary prompts to this "
	"machine, will see your OS, CPU, RAM, model list and energy telemetry, and can queue "
	"work that competes for your inference slot. You opted into this explicitly."
	)
	print(message, file=stream or sys.stderr, flush=True)
	return message


	# Retained under the old name so existing call sites keep working.
	validate_loopback_url = validate_server_url


	__all__ = [
	"ALLOW_REMOTE_SERVER_ENV_VAR",
	"BIND_HOST_ENV_VAR",
	"DEFAULT_BIND_HOST",
	"DEFAULT_SERVER_PORT",
	"DEFAULT_SERVER_URL",
	"LOOPBACK_BIND_HOST",
	"LOOPBACK_HOSTNAMES",
	"WILDCARD_BIND_HOSTS",
	"NetworkPolicyError",
	"announce_bind",
	"announce_remote_server",
	"describe_exposure",
	"is_loopback_host",
	"launch_kwargs",
	"remote_server_allowed",
	"resolve_bind_host",
	"resolve_bind_port",
	"validate_bind_host",
	"validate_loopback_url",
	"validate_server_url",
	]