interactive-story/backend/app/endpoints.py

"""Which inference endpoints this product is willing to talk to.

The Adventure Storyteller sends the player's prose, the assembled context, the
retrieved memories, and the embedding inputs to whatever address the model
endpoint names. That makes the endpoint the single most consequential setting
in the application: point it somewhere else and the whole campaign goes there.

The v1 rule (`planning/DECISIONS/002-ollama-only-v1.md`, ADR 004) is that
inference runs on user-controlled local infrastructure. Two deployments are
supported and no third is:

* **same-host** — Ollama on loopback, the default;
* **explicitly configured trusted LAN** — Ollama on another machine the user
  controls, named by them, reached over HTTP or over HTTPS with a certificate
  their machine trusts.

Everything on the public Internet is refused. Not discouraged in the UI, not
absent from a dropdown — refused, here, on the way out, so that a hand-edited
database row or a hostname that starts resolving somewhere new cannot quietly
turn a local install into an exfiltration path.

## How the line is drawn

By **address**, not by name, and against an explicit allowlist of networks:
loopback, the three RFC1918 ranges, link-local, IPv6 unique-local, and
carrier-grade NAT — the last of which is what a mesh VPN such as Tailscale
hands out and is as user-controlled as a LAN.

Every address the host resolves to must be in one of them. One address outside
is enough to refuse the endpoint, so a name resolving to both a private and a
public address does not squeak through.

The networks are spelled out rather than inferred from `ipaddress`'s own
classifications, which do not mean what this rule needs: `is_private` is true
of the documentation ranges and of `0.0.0.0/8`, and `is_reserved` is true of
IPv6 loopback — so a rule written around it refuses `http://[::1]:11434/v1`,
which is an ordinary same-host Ollama. Naming the networks keeps the policy
readable and makes anything unnamed refused by default.

Checking addresses rather than hostnames is what makes the rule hard to talk
around. A cloud provider cannot be reached by spelling its name differently,
and `localhost.` or a DNS entry pointing at a public host is judged on where it
actually goes.

## What this is not

It is not a general network-policy framework, and there is nothing to configure.
There is one predicate, and it is applied in two places: when the endpoint is
saved, so the user gets a clear error immediately, and again before every
outbound request, because a name that resolved to `192.168.1.50` this morning
can resolve to something else this afternoon.

TLS is a separate matter and is never traded against this one. See
`tlstrust.py`: certificates are verified in full, and no endpoint — however
private its address — may skip that.
"""

import ipaddress
import socket
from urllib.parse import urlparse

#: The networks an inference endpoint may live on. Anything else is refused.
ALLOWED_NETWORKS = tuple(
    ipaddress.ip_network(cidr)
    for cidr in (
        "127.0.0.0/8",      # this machine
        "10.0.0.0/8",       # RFC1918
        "172.16.0.0/12",    # RFC1918
        "192.168.0.0/16",   # RFC1918
        "169.254.0.0/16",   # link-local
        "100.64.0.0/10",    # carrier-grade NAT, which mesh VPNs use
        "::1/128",          # this machine, v6
        "fc00::/7",         # unique-local, v6
        "fe80::/10",        # link-local, v6
    )
)


def _is_local(ip) -> bool:
    return any(ip in net for net in ALLOWED_NETWORKS)

#: Hosts that are only ever a cloud inference service. The address rule below
#: already refuses every one of them, because they all resolve to public
#: addresses; this list exists solely so the error says *why* rather than
#: leaving the user to wonder whether their DNS is broken.
CLOUD_HOSTS = (
    "openrouter.ai",
    "api.openai.com",
    "api.anthropic.com",
    "api.groq.com",
    "api.mistral.ai",
    "api.together.xyz",
    "api.deepseek.com",
    "generativelanguage.googleapis.com",
    "api.cohere.ai",
    "api.perplexity.ai",
)

_CLOUD_REASON = (
    "this build talks to Ollama on your own machine or on your own network, "
    "and has no cloud provider support"
)


def _cloud_host(host: str) -> bool:
    host = host.lower().rstrip(".")
    return any(host == h or host.endswith("." + h) for h in CLOUD_HOSTS)


def rejection_reason(url: str) -> str | None:
    """Why this endpoint may not be used, or None if it may.

    The string is shown to the user, so it says what to do rather than what
    went wrong internally.
    """
    parsed = urlparse((url or "").strip())
    if parsed.scheme not in ("http", "https"):
        return "the endpoint URL must start with http:// or https://"
    host = parsed.hostname
    if not host:
        return "the endpoint URL has no host"
    if _cloud_host(host):
        return f"{host} is a cloud inference service — {_CLOUD_REASON}"

    port = parsed.port or (443 if parsed.scheme == "https" else 80)
    try:
        infos = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)
    except socket.gaierror:
        return (
            f"the host {host!r} could not be resolved — check the address, and "
            "that the machine running Ollama is reachable from here"
        )

    for info in infos:
        try:
            ip = ipaddress.ip_address(info[4][0])
        except ValueError:
            return "the endpoint host resolved to an address that could not be read"
        if _is_local(ip):
            continue
        if ip.is_global:
            return (
                f"{host} resolves to {ip}, which is a public Internet address — "
                f"{_CLOUD_REASON}. Use Ollama on this machine "
                "(http://127.0.0.1:11434/v1) or on a machine on your own network"
            )
        return (
            f"{host} resolves to {ip}, which is not on this machine and not on "
            "your own network. Use http://127.0.0.1:11434/v1, or the address of "
            "a machine on your network"
        )
    return None


def check(url: str) -> None:
    """Raises `EndpointRejected` if this endpoint is outside the policy."""
    reason = rejection_reason(url)
    if reason is not None:
        raise EndpointRejected(reason)


class EndpointRejected(Exception):
    """The configured endpoint is not one this product will send a story to."""


def is_loopback(url: str) -> bool:
    """Whether this endpoint is on this machine. Used for reporting, not for
    gating: a trusted-LAN endpoint is equally allowed."""
    host = urlparse((url or "").strip()).hostname
    if not host:
        return False
    try:
        infos = socket.getaddrinfo(host, None, type=socket.SOCK_STREAM)
    except socket.gaierror:
        return False
    try:
        return all(ipaddress.ip_address(i[4][0]).is_loopback for i in infos)
    except ValueError:
        return False