LLeMbas/src/lembas/services/agent/hosts.py

"""Whether an SSH connection is allowed to point back at this machine.

The whole design of agent chats rests on one sentence: nothing runs on the host
LLeMbas is installed on. That is why there is no local sandbox, why local MCP
over stdio is absent, and why "the security of an agent chat is the security of
the host behind its profile" is a statement anybody can check.

An SSH profile pointed at `127.0.0.1` walks straight past it. The commands go
over SSH, through a real login, and every gate in `policy.py` still applies --
and they land on the machine holding the database, the Fernet key and every
other user's encrypted credentials. Nothing else in the codebase can tell that
apart from a container on the network, because from the SSH layer's point of
view it is not different.

So it is a decision an administrator makes deliberately, in one of three
positions:

- **off** (the default, including on an instance upgrading into this) -- no
  connection may point at loopback, and one that already does is refused rather
  than quietly kept working.
- **port** -- allowed on exactly one port. This is the position that has a real
  use: a container that publishes its SSH port on the host's loopback interface
  is genuinely somewhere else, and `127.0.0.1:2222` is how you reach it. Port 22
  is refused even here, because that is the host's own sshd.
- **on** -- allowed anywhere. For somebody who has read the paragraph above and
  means it.

## Literal or resolved, and never resolved on the request path

Both are checked, at two different moments, and the split is not tidiness.

The literal forms -- `127.0.0.1`, `::1`, `localhost`, anything in
`127.0.0.0/8` -- are decided from the string with no I/O at all. That is the
check `refusal` makes, and it is why `refusal` can be called from a page render,
from `resolve_tools` and from the composer's profile listing.

A *name* that resolves to loopback needs `getaddrinfo`, which is a blocking
network call, and putting one of those behind a check that runs several times
per request is how a page render comes to wait out a DNS timeout for a host
nobody is even talking to. The first version of this file did exactly that and
the test suite went from two minutes to not finishing. So resolution happens
**only where a network call is already expected and already awaited** -- saving
a connection, and pressing Check -- and the answer is written to
`SshProfile.resolves_here`, which the request path reads for free.

The consequence, stated rather than discovered: a name whose DNS changes to
point here after it was saved is not noticed until it is saved or checked again.
That is a real gap and it is the right trade. The alternative is a DNS lookup in
front of every agent page load, and a guard that makes the application feel
broken is a guard somebody turns off.

A refusal is never silent. Every caller that has somewhere to put a sentence
puts this one there, because "this connection cannot be used" with no reason is
indistinguishable from a bug.
"""

from __future__ import annotations

import ipaddress
import logging
import socket
from typing import TYPE_CHECKING

if TYPE_CHECKING:  # pragma: no cover - typing only
    from sqlalchemy.orm import Session as DBSession

    from lembas.db.models import SshProfile

log = logging.getLogger(__name__)

MODE_OFF = "off"
MODE_PORT = "port"
MODE_ON = "on"
MODES = (MODE_OFF, MODE_PORT, MODE_ON)

MODE_LABELS = {
    MODE_OFF: "Never",
    MODE_PORT: "Only on one port",
    MODE_ON: "Anywhere",
}
MODE_HINTS = {
    MODE_OFF: (
        "A connection to this machine is refused, and an existing one stops "
        "working. This is what keeps “nothing runs on the LLeMbas host” true."
    ),
    MODE_PORT: (
        "For a container that publishes its SSH port on this machine's loopback "
        "interface. Name that port; everything else here is still refused, and "
        "port 22 is refused regardless, because that one is this host's own sshd."
    ),
    MODE_ON: (
        "Any port on this machine. Commands then run beside the database and the "
        "encryption key, with whatever the login account can reach."
    ),
}

# The host's own sshd, and never what somebody means by "the container on 2222".
HOST_SSH_PORT = 22


def _literal(host: str) -> bool | None:
    """True/False when the host decides itself, None when it needs resolving."""
    text = (host or "").strip().strip("[]").lower()
    if not text:
        return False
    # Not a real hostname anywhere, and the one everybody types.
    if text in ("localhost", "localhost.localdomain", "ip6-localhost", "ip6-loopback"):
        return True
    try:
        address = ipaddress.ip_address(text)
    except ValueError:
        return None
    # `is_unspecified` as well as `is_loopback`, because `0.0.0.0` and `::` are
    # neither a real destination nor a refused one: connect() to either goes to
    # loopback on Linux, so an SSH profile pointed at `0.0.0.0` reached this
    # host's own sshd. `is_loopback` alone answered a decided **False**, which
    # also short-circuited `resolves_here`, so the DNS half never ran either --
    # the one spelling of "this machine" that walked past a guard whose whole
    # job is that sentence.
    return address.is_loopback or address.is_unspecified


def is_loopback(host: str) -> bool:
    """Whether this host *string* reaches the machine LLeMbas is running on.

    No I/O, ever. A name is answered False here and settled by `resolves_here`
    at the two moments a lookup is affordable -- see the module docstring; the
    version of this that resolved inline made every agent page wait on DNS.
    """
    return bool(_literal(host))


def resolves_here(host: str) -> bool:
    """The same question for a name, by resolving it. Blocking; call sparingly.

    Resolution failure is answered **False**: a name that does not resolve is not
    a name pointing here, and refusing it would turn every DNS hiccup into "your
    connection is on this machine", which is both wrong and confusing. The
    connection itself will fail on its own terms a moment later.
    """
    decided = _literal(host)
    if decided is not None:
        return decided

    try:
        for entry in socket.getaddrinfo((host or "").strip().lower(), None):
            if _literal(str(entry[4][0])):
                return True
    except OSError:
        return False
    return False


def policy(db: DBSession) -> tuple[str, int]:
    """The configured position, and the port that goes with `port`."""
    from lembas.services import settings_store

    values = settings_store.agents(db)
    mode = str(values.get("loopback") or MODE_OFF)
    if mode not in MODES:
        mode = MODE_OFF
    try:
        port = int(values.get("loopback_port") or 0)
    except (TypeError, ValueError):
        port = 0
    return mode, port


def refusal(db: DBSession, host: str, port: int, *, resolved: bool = False) -> str:
    """Why this host and port may not be used, or "" if they may.

    A sentence rather than a boolean, because every caller has somewhere to show
    one and a connection that is unavailable for no stated reason reads as a
    fault in the application.

    `resolved` is what a stored profile's `resolves_here` column carries in: the
    string said nothing, and a lookup made earlier said yes.
    """
    if not (resolved or is_loopback(host)):
        return ""

    mode, allowed = policy(db)
    if mode == MODE_ON:
        return ""
    if mode == MODE_PORT:
        if allowed and port == allowed and port != HOST_SSH_PORT:
            return ""
        if allowed:
            return (
                f"This connection points at this machine, which is only allowed "
                f"on port {allowed}. An administrator sets that on the Agents page."
            )
        return (
            "This connection points at this machine, which is allowed only on a "
            "port an administrator has named — and none has been."
        )
    return (
        "This connection points at the machine LLeMbas itself runs on, which an "
        "administrator has not allowed. Agent chats are meant to reach a "
        "different host; running here would put the commands beside the database "
        "and the encryption key."
    )


def refusal_for(db: DBSession, profile: SshProfile | None) -> str:
    """The same answer for a stored profile, with no lookup.

    `resolves_here` is the verdict recorded the last time somebody saved or
    checked this connection. Reading it is what keeps this callable from a page
    render.
    """
    if profile is None:
        return ""
    return refusal(
        db, profile.host, profile.port, resolved=bool(getattr(profile, "resolves_here", False))
    )


def usable(db: DBSession, profile: SshProfile | None) -> bool:
    return not refusal_for(db, profile)


def restamp(profile: SshProfile) -> bool:
    """Record whether this profile's host resolves to loopback, and return it.

    Called where a network call is already happening -- saving a connection, and
    Check. The column is the request path's only way of knowing about a *name*,
    so a save that skips this leaves the guard reading a stale answer.
    """
    profile.resolves_here = resolves_here(profile.host)
    return profile.resolves_here


__all__ = [
    "HOST_SSH_PORT",
    "MODES",
    "MODE_HINTS",
    "MODE_LABELS",
    "MODE_OFF",
    "MODE_ON",
    "MODE_PORT",
    "is_loopback",
    "policy",
    "refusal",
    "refusal_for",
    "resolves_here",
    "restamp",
    "usable",
]