Homer
0514568df0
The testing pass: 2140 tests to 2283, and four bugs that no amount of reading had turned up. Three came from driving the JavaScript under a Node DOM stub, which is the practice CLAUDE.md sets out and this is the reason it does. The terminal dropped every keystroke after a reconnect. `onclose` closed over the module-level socket rather than its own, and close() queues its event -- so the old socket's close arrived after a new one was assigned and nulled the live one. Output kept coming, because onmessage is bound to the object, while every send gates on the variable. It also announced "Disconnected" about a shell that had just reconnected. Two scripts were loaded twice on /messages, once by base.html and again by the page. Each is an IIFE with its own state, so four keyboard shortcuts toggled their panel twice and therefore did nothing, /help opened two dialogs, and an @ mention attached its file twice. A sweep refuses any template re-loading what base.html has. The microphone had no guard while the permission prompt was up, so each click opened another stream and only the last was ever stopped. And a skill shared with you took its name out of your own library: create checked uniqueness against what is *visible* rather than what is owned, against a (owner_id, name) constraint, and told you to edit a row you cannot edit. --ink-faint failed the contrast minimum in both themes -- 3.85 and 3.19 against 4.5 -- so the smallest text on every screen was the hardest to read. Measured in a headless browser rather than judged by eye. And the suite runs on 3.11 and 3.12 now as well as 3.14. It had only ever run on 3.14 while the image ships 3.12 and the packaging claimed 3.11: the interpreter most people would run was the one nothing had tested. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
453 lines
16 KiB
Python
453 lines
16 KiB
Python
"""SSH connections, and who may do what with them.
|
|
|
|
The pages are user-facing rather than admin, because these are somebody's own
|
|
machines and somebody's own keys. Most of what is worth pinning here is about
|
|
that ownership, and about the host key -- the one thing standing between "this
|
|
is my container" and "this is something answering on its address".
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import select
|
|
|
|
from lembas.db.models import SshProfile, User
|
|
from lembas.services import settings_store
|
|
from lembas.services.crypto import UNCHANGED_SENTINEL, decrypt, encrypt
|
|
|
|
# Stands up something real -- see the `slow` marker in pyproject.toml.
|
|
pytestmark = pytest.mark.slow
|
|
|
|
asyncssh = pytest.importorskip("asyncssh")
|
|
|
|
|
|
def _form(**overrides) -> dict:
|
|
base = {
|
|
"name": "Project box",
|
|
"host": "127.0.0.1",
|
|
"port": "2222",
|
|
"username": "root",
|
|
"default_dir": "/project",
|
|
"connect_timeout": "15",
|
|
"auth": "key",
|
|
"enabled": "true",
|
|
}
|
|
base.update(overrides)
|
|
return {k: v for k, v in base.items() if v is not None}
|
|
|
|
|
|
@pytest.fixture
|
|
def ssh_host():
|
|
"""A real SSH server, on a thread and an event loop of its own.
|
|
|
|
Its own loop matters: these tests drive the app through the synchronous
|
|
TestClient, so a server sharing the test's loop could not accept a
|
|
connection while a `client.post(...)` was blocking it, and the check would
|
|
time out rather than succeed. Yields the port it is listening on.
|
|
"""
|
|
import asyncio
|
|
import threading
|
|
|
|
loop = asyncio.new_event_loop()
|
|
thread = threading.Thread(target=loop.run_forever, daemon=True)
|
|
thread.start()
|
|
|
|
async def start():
|
|
server = await asyncssh.create_server(
|
|
asyncssh.SSHServer,
|
|
"127.0.0.1",
|
|
0,
|
|
server_host_keys=[asyncssh.generate_private_key("ssh-ed25519")],
|
|
)
|
|
return server, next(iter(server.sockets)).getsockname()[1]
|
|
|
|
server, port = asyncio.run_coroutine_threadsafe(start(), loop).result(10)
|
|
try:
|
|
yield port
|
|
finally:
|
|
|
|
async def stop():
|
|
server.close()
|
|
await server.wait_closed()
|
|
|
|
asyncio.run_coroutine_threadsafe(stop(), loop).result(10)
|
|
loop.call_soon_threadsafe(loop.stop)
|
|
thread.join(timeout=5)
|
|
|
|
|
|
def _grant(db, user_id: str) -> None:
|
|
"""Give a plain account the two agent permissions."""
|
|
settings_store.update(
|
|
db, {"default_permissions": {"agent.ssh": True, "tools.agent": True}}
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def second_user(client: TestClient, db, registered):
|
|
"""Another signed-in account, so ownership can be tested."""
|
|
client.post("/auth/logout")
|
|
client.post(
|
|
"/auth/register",
|
|
data={"name": "Sam", "email": "sam@shire.test", "password": "correct horse battery"},
|
|
follow_redirects=False,
|
|
)
|
|
user = db.scalar(select(User).where(User.email == "sam@shire.test"))
|
|
user.role = "user"
|
|
user.active = True
|
|
db.commit()
|
|
_grant(db, user.id)
|
|
client.post(
|
|
"/auth/login",
|
|
data={"email": "sam@shire.test", "password": "correct horse battery"},
|
|
follow_redirects=False,
|
|
)
|
|
return user
|
|
|
|
|
|
# --- Guards --------------------------------------------------------------------
|
|
def test_the_pages_need_the_permission(client: TestClient, db, registered):
|
|
"""Administrators pass everything, so this needs a plain account."""
|
|
client.post("/auth/logout")
|
|
client.post(
|
|
"/auth/register",
|
|
data={"name": "Merry", "email": "m@shire.test", "password": "correct horse battery"},
|
|
follow_redirects=False,
|
|
)
|
|
user = db.scalar(select(User).where(User.email == "m@shire.test"))
|
|
user.role = "user"
|
|
user.active = True
|
|
db.commit()
|
|
client.post(
|
|
"/auth/login",
|
|
data={"email": "m@shire.test", "password": "correct horse battery"},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert client.get("/agents").status_code == 403
|
|
assert client.post("/api/agents", data=_form()).status_code == 403
|
|
|
|
|
|
def test_new_is_not_parsed_as_a_profile_id(client: TestClient, registered):
|
|
response = client.get("/agents/new")
|
|
assert response.status_code == 200
|
|
assert "New connection" in response.text
|
|
|
|
|
|
# --- Creating and editing ------------------------------------------------------
|
|
def test_creating_a_connection(client: TestClient, db, registered):
|
|
client.post("/api/agents", data=_form(private_key="KEY MATERIAL"), follow_redirects=False)
|
|
|
|
profile = db.scalar(select(SshProfile))
|
|
assert profile.name == "Project box"
|
|
assert profile.port == 2222
|
|
assert profile.default_dir == "/project"
|
|
assert decrypt(profile.private_key_encrypted) == "KEY MATERIAL"
|
|
assert profile.verified is False, "nothing is trusted until a key is accepted"
|
|
|
|
|
|
def test_a_duplicate_name_is_refused_per_person(client: TestClient, db, registered):
|
|
client.post("/api/agents", data=_form(), follow_redirects=False)
|
|
response = client.post("/api/agents", data=_form(), follow_redirects=False)
|
|
|
|
assert "already have a connection" in response.text
|
|
assert len(list(db.scalars(select(SshProfile)))) == 1
|
|
|
|
|
|
def test_two_people_may_use_the_same_name(client: TestClient, db, registered, second_user):
|
|
"""The uniqueness is per owner. Two people each calling theirs "box" is not
|
|
a conflict, and treating it as one would be a surprise."""
|
|
client.post("/api/agents", data=_form(name="box"), follow_redirects=False)
|
|
|
|
client.post("/auth/logout")
|
|
client.post(
|
|
"/auth/login",
|
|
data={"email": "frodo@shire.test", "password": "speak-friend-and-enter"},
|
|
follow_redirects=False,
|
|
)
|
|
client.post("/api/agents", data=_form(name="box"), follow_redirects=False)
|
|
|
|
assert len(list(db.scalars(select(SshProfile)))) == 2
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "message"),
|
|
[("name", "needs a name"), ("host", "needs a host"), ("username", "username")],
|
|
)
|
|
def test_the_essentials_are_required(client: TestClient, db, registered, field, message):
|
|
response = client.post("/api/agents", data=_form(**{field: ""}), follow_redirects=False)
|
|
assert message in response.text
|
|
assert db.scalar(select(SshProfile)) is None
|
|
|
|
|
|
def test_a_secret_is_never_rendered_in_full(client: TestClient, db, registered):
|
|
client.post("/api/agents", data=_form(private_key="SUPER SECRET KEY"), follow_redirects=False)
|
|
profile = db.scalar(select(SshProfile))
|
|
|
|
page = client.get(f"/agents/{profile.id}").text
|
|
assert "SUPER SECRET KEY" not in page
|
|
assert UNCHANGED_SENTINEL in page
|
|
|
|
|
|
def test_leaving_the_dots_alone_keeps_the_key(client: TestClient, db, registered):
|
|
client.post("/api/agents", data=_form(private_key="KEY MATERIAL"), follow_redirects=False)
|
|
profile = db.scalar(select(SshProfile))
|
|
|
|
client.post(
|
|
f"/api/agents/{profile.id}",
|
|
data=_form(private_key=UNCHANGED_SENTINEL),
|
|
follow_redirects=False,
|
|
)
|
|
db.refresh(profile)
|
|
assert decrypt(profile.private_key_encrypted) == "KEY MATERIAL"
|
|
|
|
|
|
def test_switching_to_a_password_drops_the_key(client: TestClient, db, registered):
|
|
"""Keeping a key that is no longer used would leave a credential lying in
|
|
the database with nothing pointing at it."""
|
|
client.post("/api/agents", data=_form(private_key="KEY MATERIAL"), follow_redirects=False)
|
|
profile = db.scalar(select(SshProfile))
|
|
|
|
client.post(
|
|
f"/api/agents/{profile.id}",
|
|
data=_form(auth="password", password="hunter2", private_key=UNCHANGED_SENTINEL),
|
|
follow_redirects=False,
|
|
)
|
|
db.refresh(profile)
|
|
assert profile.private_key_encrypted == ""
|
|
assert decrypt(profile.password_encrypted) == "hunter2"
|
|
|
|
|
|
# --- Ownership ------------------------------------------------------------------
|
|
def test_another_account_cannot_see_or_touch_your_connection(
|
|
client: TestClient, db, registered, second_user
|
|
):
|
|
"""`sharing.py` is deliberately not involved: it grants reading, and a host
|
|
somebody else can read is a host they can log in to."""
|
|
mine = SshProfile(
|
|
owner_id=db.scalar(select(User).where(User.email == "frodo@shire.test")).id,
|
|
name="Not yours",
|
|
host="10.0.0.5",
|
|
username="root",
|
|
)
|
|
db.add(mine)
|
|
db.commit()
|
|
|
|
# `second_user` is the one signed in.
|
|
assert client.get(f"/agents/{mine.id}").status_code == 404
|
|
assert client.post(f"/api/agents/{mine.id}", data=_form()).status_code == 404
|
|
assert client.post(f"/api/agents/{mine.id}/delete").status_code == 404
|
|
assert client.post(f"/api/agents/{mine.id}/check").status_code == 404
|
|
assert client.get("/agents").text.count("Not yours") == 0
|
|
|
|
|
|
def test_deleting_a_connection(client: TestClient, db, registered):
|
|
client.post("/api/agents", data=_form(), follow_redirects=False)
|
|
profile = db.scalar(select(SshProfile))
|
|
|
|
client.post(f"/api/agents/{profile.id}/delete", follow_redirects=False)
|
|
assert db.scalar(select(SshProfile)) is None
|
|
|
|
|
|
# --- The host key ----------------------------------------------------------------
|
|
def test_checking_an_unseen_host_offers_a_fingerprint_and_pins_nothing(
|
|
client: TestClient, db, registered, ssh_host
|
|
):
|
|
client.post("/api/agents", data=_form(port=str(ssh_host)), follow_redirects=False)
|
|
profile = db.scalar(select(SshProfile))
|
|
|
|
response = client.post(f"/api/agents/{profile.id}/check")
|
|
assert "SHA256:" in response.text
|
|
assert "Accept and pin" in response.text
|
|
|
|
db.refresh(profile)
|
|
assert profile.host_key == "", "looking is not accepting"
|
|
|
|
|
|
def test_accepting_pins_the_key_and_the_fingerprint(
|
|
client: TestClient, db, registered, ssh_host
|
|
):
|
|
client.post("/api/agents", data=_form(port=str(ssh_host)), follow_redirects=False)
|
|
profile = db.scalar(select(SshProfile))
|
|
|
|
response = client.post(f"/api/agents/{profile.id}/accept")
|
|
assert "Pinned" in response.text
|
|
|
|
db.refresh(profile)
|
|
assert profile.verified is True
|
|
assert profile.host_fingerprint.startswith("SHA256:")
|
|
assert "ssh-ed25519" in profile.host_key
|
|
|
|
|
|
def test_a_host_whose_key_changed_is_reported_and_not_silently_accepted(
|
|
client: TestClient, db, registered, ssh_host
|
|
):
|
|
client.post("/api/agents", data=_form(port=str(ssh_host)), follow_redirects=False)
|
|
profile = db.scalar(select(SshProfile))
|
|
# Pin something else entirely.
|
|
profile.host_key = f"[127.0.0.1]:{ssh_host} ssh-ed25519 {'A' * 68}\n"
|
|
profile.host_fingerprint = "SHA256:old"
|
|
db.commit()
|
|
|
|
response = client.post(f"/api/agents/{profile.id}/check")
|
|
assert "different key" in response.text
|
|
assert "Nothing was sent" in response.text
|
|
|
|
db.refresh(profile)
|
|
assert profile.host_fingerprint == "SHA256:old", "the old pin is left alone"
|
|
|
|
|
|
def test_moving_a_connection_to_another_host_forgets_its_key(client: TestClient, db, registered):
|
|
"""A pinned key belongs to a host and a port. Keeping it across a move is the
|
|
one mistake the whole mechanism exists to prevent."""
|
|
client.post("/api/agents", data=_form(), follow_redirects=False)
|
|
profile = db.scalar(select(SshProfile))
|
|
profile.host_key = "127.0.0.1 ssh-ed25519 AAAA\n"
|
|
profile.host_fingerprint = "SHA256:whatever"
|
|
db.commit()
|
|
|
|
client.post(f"/api/agents/{profile.id}", data=_form(host="10.0.0.9"), follow_redirects=False)
|
|
db.refresh(profile)
|
|
|
|
assert profile.host_key == ""
|
|
assert profile.host_fingerprint == ""
|
|
|
|
|
|
def test_editing_something_harmless_keeps_the_key(client: TestClient, db, registered):
|
|
client.post("/api/agents", data=_form(), follow_redirects=False)
|
|
profile = db.scalar(select(SshProfile))
|
|
profile.host_key = "127.0.0.1 ssh-ed25519 AAAA\n"
|
|
db.commit()
|
|
|
|
client.post(
|
|
f"/api/agents/{profile.id}", data=_form(default_dir="/elsewhere"), follow_redirects=False
|
|
)
|
|
db.refresh(profile)
|
|
|
|
assert profile.host_key == "127.0.0.1 ssh-ed25519 AAAA\n"
|
|
assert profile.default_dir == "/elsewhere"
|
|
|
|
|
|
def test_forgetting_a_key_clears_it(client: TestClient, db, registered):
|
|
client.post("/api/agents", data=_form(), follow_redirects=False)
|
|
profile = db.scalar(select(SshProfile))
|
|
profile.host_key = "127.0.0.1 ssh-ed25519 AAAA\n"
|
|
db.commit()
|
|
|
|
client.post(f"/api/agents/{profile.id}/forget")
|
|
db.refresh(profile)
|
|
assert profile.host_key == ""
|
|
|
|
|
|
# --- The admin half --------------------------------------------------------------
|
|
def test_the_admin_page_is_refused_to_a_plain_user(client: TestClient, db, registered):
|
|
client.post("/auth/logout")
|
|
client.post(
|
|
"/auth/register",
|
|
data={"name": "Pip", "email": "p@shire.test", "password": "correct horse battery"},
|
|
follow_redirects=False,
|
|
)
|
|
user = db.scalar(select(User).where(User.email == "p@shire.test"))
|
|
user.role = "user"
|
|
user.active = True
|
|
db.commit()
|
|
client.post(
|
|
"/auth/login",
|
|
data={"email": "p@shire.test", "password": "correct horse battery"},
|
|
follow_redirects=False,
|
|
)
|
|
assert client.get("/admin/agents").status_code == 403
|
|
|
|
|
|
def test_agents_are_off_until_an_administrator_says_otherwise(client: TestClient, db, registered):
|
|
assert settings_store.agents(db)["enabled"] is False
|
|
|
|
client.post(
|
|
"/admin/agents",
|
|
data={
|
|
"enabled": "true",
|
|
"default_timeout": "30",
|
|
"max_timeout": "600",
|
|
"max_output_bytes": "65536",
|
|
"max_steps": "40",
|
|
"max_wall_seconds": "900",
|
|
"max_total_output_bytes": "1048576",
|
|
"approval_timeout": "900",
|
|
"allow_default": "file_read\ngit *\n\n",
|
|
"deny_default": "shutdown *",
|
|
"ask_free_text": "true",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
values = settings_store.agents(db)
|
|
assert values["enabled"] is True
|
|
assert values["default_timeout"] == 30
|
|
assert values["allow_default"] == ["file_read", "git *"], "blank lines dropped"
|
|
assert values["deny_default"] == ["shutdown *"]
|
|
|
|
|
|
def test_the_numbers_are_clamped(client: TestClient, db, registered):
|
|
client.post(
|
|
"/admin/agents",
|
|
data={
|
|
"enabled": "true",
|
|
"default_timeout": "0",
|
|
"max_timeout": "99999",
|
|
"max_output_bytes": "1",
|
|
"max_steps": "99999",
|
|
"max_wall_seconds": "1",
|
|
"max_total_output_bytes": "1",
|
|
"max_completion_tokens": "0",
|
|
"approval_timeout": "0",
|
|
"allow_default": "",
|
|
"deny_default": "",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
values = settings_store.agents(db)
|
|
assert values["default_timeout"] == 1
|
|
assert values["max_timeout"] == 3600
|
|
assert values["max_steps"] == 1000
|
|
assert values["approval_timeout"] == 60, "a zero would park a task forever"
|
|
# Not clamped up to a minimum: zero is how "no ceiling on what a reply may
|
|
# write" is said, exactly as it is for index_chars.
|
|
assert values["max_completion_tokens"] == 0
|
|
|
|
|
|
def test_an_unticked_checkbox_turns_it_off(client: TestClient, db, registered):
|
|
settings_store.update(db, {"enabled": True}, key=settings_store.AGENTS)
|
|
client.post(
|
|
"/admin/agents",
|
|
data={
|
|
"default_timeout": "60",
|
|
"max_timeout": "600",
|
|
"max_output_bytes": "65536",
|
|
"max_steps": "40",
|
|
"max_wall_seconds": "900",
|
|
"max_total_output_bytes": "1048576",
|
|
"approval_timeout": "900",
|
|
"allow_default": "",
|
|
"deny_default": "",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
assert settings_store.agents(db)["enabled"] is False
|
|
|
|
|
|
def test_encrypted_credentials_never_appear_in_the_database_in_the_clear(db, user_id):
|
|
profile = SshProfile(
|
|
owner_id=user_id,
|
|
name="box",
|
|
host="h",
|
|
username="u",
|
|
password_encrypted=encrypt("hunter2"),
|
|
)
|
|
db.add(profile)
|
|
db.commit()
|
|
|
|
raw = db.execute(
|
|
select(SshProfile.password_encrypted).where(SshProfile.id == profile.id)
|
|
).scalar_one()
|
|
assert "hunter2" not in raw
|
|
assert decrypt(raw) == "hunter2"
|