Homer
546f8a30d7
The security pass. Six findings, none reachable by visiting the site and every one a boundary this codebase says it keeps. A subagent is pinned to a list of read-only commands, in every mode, unattended, with no card anybody could approve -- and `find *` was on it. find writes files with -fprintf, runs programs with -exec and removes them with -delete, and none of that needs a character the metacharacter guard refuses. A page the model had just read could ask for a helper and get a key into authorized_keys, from Plan mode, which promises to change nothing. Refused in `subject()` rather than trimmed from the list: a pattern cannot say "and no dangerous flags", and "this one looks read-only" is exactly what put find there. The loopback guard missed `0.0.0.0`, which is not is_loopback but does connect to localhost -- so it answered a *decided* False and skipped the DNS half too. The one spelling of "this machine" that walked past a guard whose whole job is that sentence. Twice in the update helper, which is the one place this deliberately crosses a privilege boundary: root ran a script the service account owns, and root sourced a file that account can replace. Either turns a compromise of the web application into root. The first needed no compromise at all -- a pull happens as the service user and root runs whatever it fetched, so control of the branch was control of root. The old test asserted that exact ExecStart line and had pinned it in place. Push endpoints skipped check_url, the only outbound request that did. And a chat could be filed in another account's folder, which hands over its system prompt -- `_new_chat` resolved the folder, discarded it when it was not the caller's, and stored the raw id anyway. An existing helper install keeps the old wiring until install.sh is re-run; update.sh now says so when it finds itself inside the checkout. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
289 lines
10 KiB
Python
289 lines
10 KiB
Python
"""An SSH connection pointed at the machine LLeMbas itself runs on.
|
|
|
|
The whole design of agent chats is one sentence -- nothing runs on this host --
|
|
and a profile pointed at 127.0.0.1 walks past it while looking, from the SSH
|
|
layer down, exactly like a container on the network. So it is an administrator's
|
|
decision in three positions, and what is pinned here is that every one of them is
|
|
enforced where it has to be rather than only in the form.
|
|
|
|
Note the conftest: the suite runs with the switch **open**, because the tests
|
|
that stand up a real asyncssh server can only listen on loopback. Every test in
|
|
this file closes it explicitly, which is also why they are together in one file.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import select
|
|
|
|
from lembas.db.models import KIND_AGENT, Chat, SshProfile, User
|
|
from lembas.services import settings_store
|
|
from lembas.services.agent import hosts
|
|
from lembas.services.agent import session as agent_session
|
|
from lembas.services.crypto import encrypt
|
|
|
|
|
|
def _set(db, mode: str, port: int = 0) -> None:
|
|
settings_store.update(
|
|
db, {"enabled": True, "loopback": mode, "loopback_port": port}, key=settings_store.AGENTS
|
|
)
|
|
db.commit()
|
|
|
|
|
|
def _profile(db, host: str = "127.0.0.1", port: int = 22) -> SshProfile:
|
|
profile = SshProfile(
|
|
owner_id=db.scalars(select(User)).first().id,
|
|
name=f"box-{host}-{port}",
|
|
host=host,
|
|
port=port,
|
|
username="deploy",
|
|
auth="password",
|
|
password_encrypted=encrypt("hunter2"),
|
|
host_key="ssh-ed25519 AAAA",
|
|
host_fingerprint="SHA256:x",
|
|
enabled=True,
|
|
)
|
|
db.add(profile)
|
|
db.commit()
|
|
return profile
|
|
|
|
|
|
# --- Recognising the machine --------------------------------------------------
|
|
@pytest.mark.parametrize(
|
|
"host",
|
|
["127.0.0.1", "127.0.0.53", "127.1.2.3", "::1", "[::1]", "localhost", "LOCALHOST"],
|
|
)
|
|
def test_the_obvious_spellings_are_recognised(host: str):
|
|
"""All of them, and from the string alone. `127.0.0.53` is the one worth
|
|
listing: the whole of 127.0.0.0/8 is loopback, and a check written against
|
|
the single address `127.0.0.1` misses every other way of writing it."""
|
|
assert hosts.is_loopback(host)
|
|
|
|
|
|
@pytest.mark.parametrize("host", ["10.0.0.9", "192.168.1.5", "example.test", ""])
|
|
def test_anything_else_is_not(host: str):
|
|
"""Including a private address. This guard is about *this machine*, not
|
|
about the network it is on -- refusing 192.168.x would refuse the container
|
|
on the shelf, which is the intended use of the whole feature."""
|
|
assert not hosts.is_loopback(host)
|
|
|
|
|
|
def test_recognising_a_host_never_makes_a_network_call(monkeypatch):
|
|
"""`getaddrinfo` blocks. `is_loopback` is called several times per page
|
|
render -- the composer's picker, `resolve_tools`, the canvas -- so a lookup
|
|
behind it makes an agent page wait out a DNS timeout for a host nobody is
|
|
talking to. The first version of this did exactly that and the suite went
|
|
from two minutes to not finishing."""
|
|
import socket
|
|
|
|
def explode(*args, **kwargs): # pragma: no cover - the point is it is unused
|
|
raise AssertionError("is_loopback resolved a name on the request path")
|
|
|
|
monkeypatch.setattr(socket, "getaddrinfo", explode)
|
|
|
|
assert hosts.is_loopback("127.0.0.1")
|
|
assert not hosts.is_loopback("build.example")
|
|
|
|
|
|
def test_a_name_is_settled_by_resolving_it(monkeypatch):
|
|
"""Which is where a hosts-file entry gets caught. Blocking, and therefore
|
|
called only where a network call is already expected."""
|
|
import socket
|
|
|
|
monkeypatch.setattr(
|
|
socket,
|
|
"getaddrinfo",
|
|
lambda *a, **k: [(2, 1, 6, "", ("127.0.0.1", 22))],
|
|
)
|
|
assert hosts.resolves_here("build.example")
|
|
|
|
monkeypatch.setattr(
|
|
socket,
|
|
"getaddrinfo",
|
|
lambda *a, **k: [(2, 1, 6, "", ("93.184.216.34", 22))],
|
|
)
|
|
assert not hosts.resolves_here("build.example")
|
|
|
|
|
|
def test_a_name_that_does_not_resolve_is_not_this_machine(monkeypatch):
|
|
"""Refusing it would turn every DNS hiccup into "your connection is on this
|
|
machine", which is both wrong and unactionable. The connection fails on its
|
|
own terms a moment later."""
|
|
import socket
|
|
|
|
def fail(*args, **kwargs):
|
|
raise OSError("no such host")
|
|
|
|
monkeypatch.setattr(socket, "getaddrinfo", fail)
|
|
assert not hosts.resolves_here("nowhere.invalid")
|
|
|
|
|
|
# --- The three positions ------------------------------------------------------
|
|
def test_off_refuses_every_port(db, registered):
|
|
_set(db, hosts.MODE_OFF)
|
|
assert hosts.refusal(db, "127.0.0.1", 22)
|
|
assert hosts.refusal(db, "127.0.0.1", 2222)
|
|
|
|
|
|
def test_on_allows_every_port(db, registered):
|
|
_set(db, hosts.MODE_ON)
|
|
assert hosts.refusal(db, "127.0.0.1", 22) == ""
|
|
assert hosts.refusal(db, "127.0.0.1", 2222) == ""
|
|
|
|
|
|
def test_one_port_allows_that_one_and_nothing_else(db, registered):
|
|
_set(db, hosts.MODE_PORT, 2222)
|
|
assert hosts.refusal(db, "127.0.0.1", 2222) == ""
|
|
assert hosts.refusal(db, "127.0.0.1", 2200)
|
|
# And a host that is not this machine is not this switch's business.
|
|
assert hosts.refusal(db, "10.0.0.9", 22) == ""
|
|
|
|
|
|
def test_port_22_is_refused_even_when_it_is_the_named_one(db, registered):
|
|
"""The position exists for a container that published its SSH port on the
|
|
loopback interface. Port 22 is not that -- it is this host's own sshd, and
|
|
naming it is either a mistake or the thing the switch exists to stop."""
|
|
_set(db, hosts.MODE_PORT, 22)
|
|
assert hosts.refusal(db, "127.0.0.1", 22)
|
|
|
|
|
|
def test_naming_no_port_refuses_rather_than_allows(db, registered):
|
|
"""`port` with nothing named is an unfinished setting, and the safe reading
|
|
of an unfinished setting is the one that refuses."""
|
|
_set(db, hosts.MODE_PORT, 0)
|
|
assert hosts.refusal(db, "127.0.0.1", 2222)
|
|
|
|
|
|
def test_an_unrecognised_stored_position_means_off(db, registered):
|
|
"""Fail closed. The direction that is safe to get wrong is refusing a
|
|
connection somebody has to re-allow; the other direction is a shell on the
|
|
host holding the database and the encryption key."""
|
|
settings_store.update(db, {"enabled": True, "loopback": "sometimes"}, key=settings_store.AGENTS)
|
|
db.commit()
|
|
assert hosts.refusal(db, "127.0.0.1", 2222)
|
|
|
|
|
|
# --- Where it is enforced -----------------------------------------------------
|
|
def test_a_refused_connection_cannot_be_saved(client: TestClient, db, registered):
|
|
_set(db, hosts.MODE_OFF)
|
|
|
|
response = client.post(
|
|
"/api/agents",
|
|
data={
|
|
"name": "Here",
|
|
"host": "127.0.0.1",
|
|
"port": "22",
|
|
"username": "deploy",
|
|
"auth": "password",
|
|
"password": "hunter2",
|
|
"enabled": "on",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == 200 # the form again, with the reason on it
|
|
assert "points at the machine LLeMbas itself runs on" in response.text
|
|
assert db.scalar(select(SshProfile).where(SshProfile.name == "Here")) is None
|
|
|
|
|
|
def test_a_connection_saved_before_the_switch_moved_stops_working(db, registered):
|
|
"""The one that matters. A row can predate a setting, so refusing only at
|
|
save time would leave every existing loopback profile working exactly as it
|
|
did -- which is the whole of the flaw, untouched.
|
|
|
|
`session.resolve` is the control: it is what every agent tool, the terminal
|
|
and the canvas go through.
|
|
"""
|
|
_set(db, hosts.MODE_ON)
|
|
profile = _profile(db)
|
|
chat = Chat(
|
|
user_id=profile.owner_id, kind=KIND_AGENT, ssh_profile_id=profile.id, title="t"
|
|
)
|
|
db.add(chat)
|
|
db.commit()
|
|
|
|
owner = db.get(User, profile.owner_id)
|
|
assert agent_session.resolve(db, chat, owner) is not None
|
|
|
|
_set(db, hosts.MODE_OFF)
|
|
assert agent_session.resolve(db, chat, owner) is None
|
|
|
|
|
|
def test_a_refused_connection_is_not_offered_in_the_composer(
|
|
client: TestClient, db, registered
|
|
):
|
|
"""Or the reader picks it, starts an agent chat, and finds a chat with no
|
|
tools and nothing said about why."""
|
|
_set(db, hosts.MODE_ON)
|
|
profile = _profile(db)
|
|
|
|
assert profile.name in client.get("/chat").text
|
|
|
|
_set(db, hosts.MODE_OFF)
|
|
assert profile.name not in client.get("/chat").text
|
|
|
|
|
|
def test_the_listing_says_why_rather_than_hiding_it(client: TestClient, db, registered):
|
|
"""The connection is still the owner's and still editable -- pointing it at
|
|
another host is the way out. A row that silently stopped working would be
|
|
indistinguishable from a fault."""
|
|
_set(db, hosts.MODE_OFF)
|
|
profile = _profile(db)
|
|
|
|
body = client.get("/agents").text
|
|
|
|
assert profile.name in body
|
|
assert "not allowed" in body
|
|
assert "points at the machine LLeMbas itself runs on" in body
|
|
|
|
|
|
def test_check_opens_no_socket_for_a_refused_connection(
|
|
client: TestClient, db, registered, monkeypatch
|
|
):
|
|
"""Check is the one button here that connects, so it must refuse before it
|
|
does rather than after."""
|
|
_set(db, hosts.MODE_OFF)
|
|
profile = _profile(db)
|
|
|
|
async def explode(*args, **kwargs): # pragma: no cover - the point is it is unused
|
|
raise AssertionError("a refused connection was contacted")
|
|
|
|
from lembas.services.agent import ssh as ssh_service
|
|
|
|
monkeypatch.setattr(ssh_service, "capture_host_key", explode)
|
|
|
|
body = client.post(f"/api/agents/{profile.id}/check").text
|
|
|
|
assert "points at the machine LLeMbas itself runs on" in body
|
|
|
|
|
|
def test_a_draft_cannot_be_opened_against_a_refused_connection(
|
|
client: TestClient, db, registered
|
|
):
|
|
"""The draft is what the terminal and the canvas open against before a chat
|
|
exists, so this is the new-chat path's own gate."""
|
|
_set(db, hosts.MODE_OFF)
|
|
profile = _profile(db)
|
|
|
|
assert client.get(f"/api/agents/{profile.id}/draft").status_code == 403
|
|
|
|
|
|
def test_the_unspecified_address_is_this_machine_too():
|
|
"""`0.0.0.0` and `::` are neither a real destination nor a refused one:
|
|
connect() to either goes to loopback on Linux, so a 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, and the one spelling of
|
|
"this machine" that mattered walked past a guard whose whole job is that
|
|
sentence.
|
|
"""
|
|
for spelling in ("0.0.0.0", "::", "[::]", "0.0.0.0 "):
|
|
assert hosts.is_loopback(spelling), spelling
|
|
|
|
|
|
def test_a_real_address_is_still_not_this_machine():
|
|
for spelling in ("192.168.1.5", "10.0.0.1", "example.com", "203.0.113.9"):
|
|
assert not hosts.is_loopback(spelling), spelling
|