Houmeres/LLeMbas
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

SSH connections, kept by the people who own them

Browse Source 
An agent chat will act on a machine you choose, so this is the screen where
you choose it. User-owned like a note, not admin-owned like a connection:
these are somebody's own machines and somebody's own keys, and "anyone in
this group may log in to my server" is a different feature with a different
blast radius. services/sharing.py is deliberately not involved either --
sharing grants reading, and a host somebody else can read is a host they
can log in to.

Trust on first use, made explicit rather than assumed. Adding a host does
not connect to it. Check looks at its key and shows you the fingerprint;
nothing is sent until you accept, because get_server_host_key completes the
key exchange and stops -- no username, no credential. Accepting pins it,
and a host that later presents a different key is refused with the reason
rather than quietly trusted. Moving a profile to another host or port
forgets the pin, since a key belongs to the machine it came from.

Four asyncssh defaults are actively wrong here and all four are passed
explicitly: every LLeMbas user shares one unix account, so `known_hosts`
would be a shared trust store, `client_keys` would authenticate one person
with another's key, `config` would let a ProxyCommand redirect the
connection, and `agent_path` would silently use $SSH_AUTH_SOCK. There is a
test for exactly that, and it needs no server.

Files go over SFTP rather than through a shell. The SSH exec protocol
carries one command *string* that the far side parses, with no argv form at
all, so a model-supplied path in a command line is unavoidably a quoting
problem. Over SFTP a path is a path.

Chat gains its kind, connection, project directory and mode; the first
three are fixed once a chat has a message, because a transcript whose
earlier turns ran somewhere else is not one conversation.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jaroslav Beneš
2026-08-01 22:34:58 +02:00
parent 0a4531f02d
commit 4ced049ff8
21 changed files with 2452 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_agent_profiles.py
+445
View File
@@ -0,0 +1,445 @@
"""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
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": "9999",
"max_wall_seconds": "1",
"max_total_output_bytes": "1",
"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"] == 200
assert values["approval_timeout"] == 60, "a zero would park a task forever"
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"
tests/test_agent_ssh.py
+327
View File
@@ -0,0 +1,327 @@
"""Acting on a machine over SSH.
Driven against a real asyncssh server on 127.0.0.1 with a generated host key, so
nothing here touches an outside network and the host-key path is exercised for
real rather than mocked.
"""
from __future__ import annotations
import asyncio
import pytest
from lembas.services.agent import ssh
from lembas.services.agent.base import ExecError, ExecRequest
asyncssh = pytest.importorskip("asyncssh")
# --- A server to talk to -------------------------------------------------------
def _generate_key():
return asyncssh.generate_private_key("ssh-ed25519")
class _Server(asyncssh.SSHServer):
def begin_auth(self, username: str) -> bool:
return False # no authentication wanted; every login succeeds
async def _handler(process):
"""A shell that understands just enough to be tested against."""
command = process.command or ""
if "sleep" in command:
await asyncio.sleep(5)
process.exit(0)
return
if "flood" in command:
process.stdout.write("x" * 200_000)
process.exit(0)
return
if "fail" in command:
process.stderr.write("it went wrong\n")
process.exit(3)
return
if "colour" in command:
process.stdout.write("\x1b[31mred\x1b[0m\n")
process.exit(0)
return
process.stdout.write(f"ran: {command}\n")
process.exit(0)
async def _start(host_key=None):
server = await asyncssh.create_server(
_Server,
"127.0.0.1",
0,
server_host_keys=[host_key or _generate_key()],
process_factory=_handler,
sftp_factory=True,
)
port = next(iter(server.sockets)).getsockname()[1]
return server, port
def _spec(port: int, host_key: str, **overrides) -> dict:
base = {
"id": "p1",
"label": "test box",
"host": "127.0.0.1",
"port": port,
"username": "tester",
"auth": "key",
"password": "",
"private_key": "",
"key_passphrase": "",
"host_key": host_key,
"connect_timeout": 5,
}
base.update(overrides)
return base
@pytest.fixture
async def project(tmp_path):
"""A directory of its own.
Not `tmp_path` itself: the autouse database fixture points the data
directory there, so a listing would find lembas.db and the uploads folder.
"""
path = tmp_path / "project"
path.mkdir()
return path
@pytest.fixture
async def box(project):
"""A running server, its pinned host key line, and a project directory."""
key = _generate_key()
server, port = await _start(key)
line, fingerprint = await ssh.capture_host_key("127.0.0.1", port)
try:
yield {
"port": port,
"host_key": line,
"fingerprint": fingerprint,
"dir": str(project),
"server": server,
}
finally:
server.close()
await server.wait_closed()
# --- The four defaults that must never be left to asyncssh --------------------
def test_the_dangerous_defaults_are_all_passed_explicitly():
"""Every LLeMbas user shares one unix account, so "whatever the account has
lying around" is never the right answer. This is the most important test in
the file and it needs no server at all."""
kwargs = ssh._connect_kwargs(_spec(22, "127.0.0.1 ssh-ed25519 AAAA\n"))
# Bytes, never None: None turns host key checking off entirely.
assert isinstance(kwargs["known_hosts"], bytes)
assert kwargs["known_hosts"] != b""
# Not left to load ~/.ssh/id_*, which could be another person's key.
assert kwargs["client_keys"] == []
# Not left to read ~/.ssh/config, where ProxyCommand could redirect us.
assert kwargs["config"] is None
# Not left to use $SSH_AUTH_SOCK.
assert kwargs["agent_path"] is None
def test_a_profile_with_no_confirmed_host_key_refuses_to_connect():
with pytest.raises(ExecError, match="host key has not been confirmed"):
ssh._connect_kwargs(_spec(22, ""))
def test_a_password_profile_sends_a_password_and_no_keys():
kwargs = ssh._connect_kwargs(
_spec(22, "h ssh-ed25519 AAAA\n", auth="password", password="hunter2")
)
assert kwargs["password"] == "hunter2"
assert kwargs["client_keys"] == []
def test_a_key_profile_sends_no_password():
kwargs = ssh._connect_kwargs(_spec(22, "h ssh-ed25519 AAAA\n", password="stale"))
assert kwargs["password"] is None
# --- Trust on first use --------------------------------------------------------
async def test_the_first_look_captures_a_key_and_a_fingerprint(box):
assert box["host_key"].startswith(f"[127.0.0.1]:{box['port']} ssh-ed25519 ")
assert box["fingerprint"].startswith("SHA256:")
async def test_capturing_a_key_offers_no_credential():
"""get_server_host_key completes the key exchange and stops, which is what
makes accepting a fingerprint from a button safe: nothing is sent to a host
that has not been accepted yet.
The server here records every authentication attempt, so an empty list is
evidence rather than an absence of it.
"""
attempts: list[str] = []
class Watchful(asyncssh.SSHServer):
def begin_auth(self, username: str) -> bool:
attempts.append(username)
return False
server = await asyncssh.create_server(
Watchful, "127.0.0.1", 0, server_host_keys=[_generate_key()]
)
port = next(iter(server.sockets)).getsockname()[1]
try:
line, fingerprint = await ssh.capture_host_key("127.0.0.1", port)
assert line and fingerprint.startswith("SHA256:")
assert attempts == [], "a host not yet accepted was offered a username"
finally:
server.close()
await server.wait_closed()
async def test_a_host_that_answers_with_a_different_key_is_refused(box):
"""The pinned key is the whole of the protection. A host presenting another
one is either rebuilt or is not the host."""
other_line, _ = await ssh.capture_host_key("127.0.0.1", box["port"])
wrong = other_line.rsplit(" ", 1)[0] + " " + "A" * 68 + "\n"
executor = ssh.SshExecutor(_spec(box["port"], wrong), box["dir"])
with pytest.raises(ExecError):
await executor.run(ExecRequest(command="echo hi", timeout=5))
async def test_an_unreachable_host_says_so():
with pytest.raises(ExecError, match="Could not reach"):
await ssh.capture_host_key("127.0.0.1", 1, timeout=3)
# --- Running commands ----------------------------------------------------------
async def test_a_command_runs_and_comes_back(box):
executor = ssh.SshExecutor(_spec(box["port"], box["host_key"]), box["dir"])
result = await executor.run(ExecRequest(command="echo hi", timeout=5))
assert result.ok
assert result.exit_status == 0
assert "echo hi" in result.output
async def test_the_working_directory_is_set_and_never_spliced_in(box):
"""Every command is a fresh shell, so `cd` cannot carry between calls. The
directory is quoted, because `cwd` may come from the model."""
executor = ssh.SshExecutor(_spec(box["port"], box["host_key"]), "/tmp/a dir")
result = await executor.run(ExecRequest(command="echo hi", timeout=5))
assert "cd '/tmp/a dir' && echo hi" in result.output
async def test_a_directory_with_a_quote_in_it_cannot_end_the_quoting(box):
executor = ssh.SshExecutor(_spec(box["port"], box["host_key"]), "/tmp/it's; rm -rf /")
result = await executor.run(ExecRequest(command="echo hi", timeout=5))
assert "'/tmp/it'\\''s; rm -rf /'" in result.output
async def test_a_failing_command_is_a_result_not_an_error(box):
"""A command that ran and failed is something the model should read and
react to. Only being unable to act at all is an ExecError."""
executor = ssh.SshExecutor(_spec(box["port"], box["host_key"]), box["dir"])
result = await executor.run(ExecRequest(command="fail", timeout=5))
assert result.exit_status == 3
assert result.ok is False
assert "it went wrong" in result.output, "stderr is interleaved, not dropped"
async def test_a_slow_command_times_out_and_says_so(box):
executor = ssh.SshExecutor(_spec(box["port"], box["host_key"]), box["dir"])
result = await executor.run(ExecRequest(command="sleep", timeout=1))
assert result.timed_out is True
assert result.ok is False
assert "was stopped" in result.output
async def test_a_flood_of_output_is_capped(box):
executor = ssh.SshExecutor(_spec(box["port"], box["host_key"]), box["dir"])
result = await executor.run(ExecRequest(command="flood", timeout=10, max_bytes=2000))
assert result.truncated is True
assert len(result.output) < 2200
assert result.output.endswith("(truncated)")
async def test_terminal_escapes_are_stripped(box):
"""Inert in escaped HTML, but this text re-enters the model's context, where
they are a known way to hide instructions -- and a log a person later cats."""
executor = ssh.SshExecutor(_spec(box["port"], box["host_key"]), box["dir"])
result = await executor.run(ExecRequest(command="colour", timeout=5))
assert "red" in result.output
assert "\x1b" not in result.output
# --- Files, over SFTP ----------------------------------------------------------
async def test_a_file_is_written_and_read_back(box):
executor = ssh.SshExecutor(_spec(box["port"], box["host_key"]), box["dir"])
written = await executor.write_file("hello.txt", "a mallorn tree\n")
assert written == len("a mallorn tree\n")
assert await executor.read_file("hello.txt") == "a mallorn tree\n"
async def test_a_relative_path_resolves_against_the_project_directory(box, project):
executor = ssh.SshExecutor(_spec(box["port"], box["host_key"]), box["dir"])
await executor.write_file("nested.txt", "here")
assert (project / "nested.txt").read_text() == "here"
async def test_a_missing_file_is_reported_plainly(box):
executor = ssh.SshExecutor(_spec(box["port"], box["host_key"]), box["dir"])
with pytest.raises(ExecError, match="no file at"):
await executor.read_file("nope.txt")
async def test_a_directory_lists(box, project):
(project / "one.txt").write_text("1")
(project / "two.txt").write_text("2")
executor = ssh.SshExecutor(_spec(box["port"], box["host_key"]), box["dir"])
assert await executor.list_dir() == ["one.txt", "two.txt"]
async def test_reading_a_file_is_capped(box, project):
(project / "big.txt").write_text("y" * 50_000)
executor = ssh.SshExecutor(_spec(box["port"], box["host_key"]), box["dir"])
text = await executor.read_file("big.txt", max_bytes=1000)
assert len(text) <= 1000
# --- The check a person presses ------------------------------------------------
async def test_check_reports_what_it_found(box):
found = await ssh.check(_spec(box["port"], box["host_key"]), box["dir"])
assert found["ok"] is True
assert "uname" in found["output"]
# --- The snapshot --------------------------------------------------------------
def test_spec_from_decrypts_the_credential_and_the_row_does_not_hold_it(db, user_id):
from lembas.db.models import SshProfile
from lembas.services.crypto import encrypt
profile = SshProfile(
owner_id=user_id,
name="box",
host="10.0.0.5",
username="root",
private_key_encrypted=encrypt("PRIVATE KEY MATERIAL"),
host_key="h ssh-ed25519 AAAA\n",
)
db.add(profile)
db.commit()
spec = ssh.spec_from(profile)
assert spec["private_key"] == "PRIVATE KEY MATERIAL"
assert "PRIVATE KEY MATERIAL" not in profile.private_key_encrypted