Jaroslav Beneš
803d808723
The mode select in the topbar posted with hx-post against a route that only answers PATCH, so every change returned 405 and the mode never moved. htmx shows nothing when a request fails, so the control looked like it worked: the select stayed where you put it and the server ignored you. It has never worked. Two more of the same kind. A mode could not be chosen at all until the chat existed, so reaching Plan meant sending something in Manual first and letting the model answer under the wrong rules. And the project directory box was real and submitted, but unlabelled and squeezed to a few characters by the select beside it, so it read as broken -- which is how it was reported. So the kind, the connection, the directory and the mode move out of the strip above the text and into one toolbar row beneath it, where attach and send already are. The directory becomes a button that opens a browser over SFTP, because a path is something you would rather find than spell. `scan_dir` is new beside `list_dir`: a picker has to tell a directory from a file before it can draw the row, and `list_dir` backs a tool whose contract is a list of names and must not change under a model mid-conversation. Browsing is a person clicking, not a model calling, so it does not pass through policy.py -- the same argument the terminal panel rests on. It does mean Manual mode has a second exception now. Also: .chip was two components with one name, and the attachment card won, so the Chat/Agent pills silently wore its padding. --radius-md was used twice and declared nowhere, so both fell back to 0. .btn.is-active has been set by syncToggles since the terminal landed and styled by nothing. Enter-to-send ignored isComposing, so committing an IME candidate sent the message. The terminal had five colours of a sixteen-colour palette, with fallbacks from a palette that no longer exists. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
257 lines
8.8 KiB
Python
257 lines
8.8 KiB
Python
"""Walking a directory on the far side, to choose where a chat works.
|
|
|
|
Everything here goes through a real SFTP server on 127.0.0.1. A stub would
|
|
prove nothing worth proving: the whole point of `scan_dir` is that it reads
|
|
`SFTPName.attrs` to tell a directory from a file, and a stub that returns
|
|
whatever shape the test wants cannot be wrong about that.
|
|
"""
|
|
|
|
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.agent import ssh as ssh_service
|
|
|
|
asyncssh = pytest.importorskip("asyncssh")
|
|
|
|
|
|
class _Server(asyncssh.SSHServer):
|
|
def begin_auth(self, username: str) -> bool:
|
|
return False
|
|
|
|
|
|
@pytest.fixture
|
|
async def tree(tmp_path):
|
|
"""A real sshd with SFTP, over a directory laid out to be walked."""
|
|
root = tmp_path / "project"
|
|
(root / "src" / "deep").mkdir(parents=True)
|
|
(root / "docs").mkdir()
|
|
(root / "README.md").write_text("hello")
|
|
(root / "src" / "main.py").write_text("print()")
|
|
(root / ".hidden").mkdir()
|
|
|
|
server = await asyncssh.create_server(
|
|
_Server,
|
|
"127.0.0.1",
|
|
0,
|
|
server_host_keys=[asyncssh.generate_private_key("ssh-ed25519")],
|
|
sftp_factory=True,
|
|
)
|
|
port = next(iter(server.sockets)).getsockname()[1]
|
|
line, fingerprint = await ssh_service.capture_host_key("127.0.0.1", port)
|
|
try:
|
|
yield {"port": port, "host_key": line, "fingerprint": fingerprint, "root": str(root)}
|
|
finally:
|
|
server.close()
|
|
await server.wait_closed()
|
|
|
|
|
|
def _spec(tree: dict) -> dict:
|
|
return {
|
|
"host": "127.0.0.1",
|
|
"port": tree["port"],
|
|
"username": "tester",
|
|
"auth": "key",
|
|
"password": "",
|
|
"private_key": "",
|
|
"key_passphrase": "",
|
|
"host_key": tree["host_key"],
|
|
"connect_timeout": 15,
|
|
}
|
|
|
|
|
|
# --- scan_dir ----------------------------------------------------------------
|
|
async def test_a_listing_says_which_rows_are_directories(tree):
|
|
"""The whole reason this exists beside `list_dir`, which returns names."""
|
|
executor = ssh_service.SshExecutor(_spec(tree), tree["root"])
|
|
entries = await executor.scan_dir("")
|
|
|
|
kinds = {entry.name: entry.is_dir for entry in entries}
|
|
assert kinds["src"] is True
|
|
assert kinds["docs"] is True
|
|
assert kinds["README.md"] is False
|
|
|
|
|
|
async def test_directories_sort_before_files(tree):
|
|
"""The order somebody navigating expects: the things you can walk into."""
|
|
executor = ssh_service.SshExecutor(_spec(tree), tree["root"])
|
|
names = [entry.name for entry in await executor.scan_dir("")]
|
|
|
|
assert names.index("src") < names.index("README.md")
|
|
assert names.index("docs") < names.index("README.md")
|
|
|
|
|
|
async def test_files_carry_a_size(tree):
|
|
executor = ssh_service.SshExecutor(_spec(tree), tree["root"])
|
|
entries = {entry.name: entry for entry in await executor.scan_dir("")}
|
|
|
|
assert entries["README.md"].size == len("hello")
|
|
|
|
|
|
async def test_dotfiles_are_listed(tree):
|
|
"""Shown, not filtered. A picker that hides `.config` is one somebody has
|
|
to work around, and the account on the far side is the boundary here --
|
|
not a taste for tidy listings."""
|
|
executor = ssh_service.SshExecutor(_spec(tree), tree["root"])
|
|
names = [entry.name for entry in await executor.scan_dir("")]
|
|
|
|
assert ".hidden" in names
|
|
assert "." not in names and ".." not in names
|
|
|
|
|
|
async def test_a_relative_path_is_measured_from_the_project_directory(tree):
|
|
executor = ssh_service.SshExecutor(_spec(tree), tree["root"])
|
|
names = [entry.name for entry in await executor.scan_dir("src")]
|
|
|
|
assert names == ["deep", "main.py"]
|
|
|
|
|
|
async def test_an_absolute_path_walks_out_of_the_project_directory(tree):
|
|
"""Deliberate, and matching `_resolve`'s own docstring.
|
|
|
|
Containment is the far-side account's job, not this layer's: `shell_run`
|
|
could leave the project directory in one line, so a picker that refused to
|
|
would be a comfort rather than a control. The browser opens *at* the
|
|
project directory; it does not fence it.
|
|
"""
|
|
executor = ssh_service.SshExecutor(_spec(tree), tree["root"])
|
|
names = [entry.name for entry in await executor.scan_dir("/")]
|
|
|
|
assert names # the real filesystem root, listed without complaint
|
|
|
|
|
|
async def test_a_missing_directory_says_so_in_words(tree):
|
|
executor = ssh_service.SshExecutor(_spec(tree), tree["root"])
|
|
|
|
with pytest.raises(ssh_service.ExecError) as caught:
|
|
await executor.scan_dir("nowhere")
|
|
|
|
assert "no directory" in caught.value.message.lower()
|
|
|
|
|
|
# --- The route ---------------------------------------------------------------
|
|
@pytest.fixture
|
|
def served_tree(tmp_path):
|
|
"""The same tree, but with the server on a thread and a loop of its own.
|
|
|
|
Its own loop matters: the tests below drive the app through the synchronous
|
|
TestClient, so a server sharing the test's loop could never accept the
|
|
connection the route is trying to make -- `client.get(...)` is still
|
|
blocking it. The async fixture above is fine for calling `scan_dir`
|
|
directly, and useless here.
|
|
"""
|
|
import asyncio
|
|
import threading
|
|
|
|
root = tmp_path / "served"
|
|
(root / "src").mkdir(parents=True)
|
|
(root / "README.md").write_text("hello")
|
|
|
|
loop = asyncio.new_event_loop()
|
|
thread = threading.Thread(target=loop.run_forever, daemon=True)
|
|
thread.start()
|
|
|
|
async def start():
|
|
server = await asyncssh.create_server(
|
|
_Server,
|
|
"127.0.0.1",
|
|
0,
|
|
server_host_keys=[asyncssh.generate_private_key("ssh-ed25519")],
|
|
sftp_factory=True,
|
|
)
|
|
port = next(iter(server.sockets)).getsockname()[1]
|
|
line, fingerprint = await ssh_service.capture_host_key("127.0.0.1", port)
|
|
return server, port, line, fingerprint
|
|
|
|
server, port, line, fingerprint = asyncio.run_coroutine_threadsafe(start(), loop).result(10)
|
|
try:
|
|
yield {"port": port, "host_key": line, "fingerprint": fingerprint, "root": str(root)}
|
|
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 _profile(db, tree: dict, *, host_key: str | None = None) -> SshProfile:
|
|
user = db.scalars(select(User)).first()
|
|
profile = SshProfile(
|
|
owner_id=user.id,
|
|
name="Test box",
|
|
host="127.0.0.1",
|
|
port=tree["port"],
|
|
username="tester",
|
|
host_key=tree["host_key"] if host_key is None else host_key,
|
|
host_fingerprint=tree["fingerprint"],
|
|
default_dir=tree["root"],
|
|
)
|
|
db.add(profile)
|
|
db.commit()
|
|
return profile
|
|
|
|
|
|
def test_browsing_lists_the_project_directory(client: TestClient, db, registered, served_tree):
|
|
profile = _profile(db, served_tree)
|
|
|
|
body = client.get(f"/api/agents/{profile.id}/browse").text
|
|
|
|
assert "src" in body
|
|
assert "README.md" in body
|
|
assert served_tree["root"] in body
|
|
|
|
|
|
def test_browsing_offers_a_way_back_up(client: TestClient, db, registered, served_tree):
|
|
profile = _profile(db, served_tree)
|
|
|
|
body = client.get(f"/api/agents/{profile.id}/browse", params={"path": served_tree["root"]}).text
|
|
|
|
assert "Up a level" in body
|
|
|
|
|
|
def test_a_host_key_that_was_never_confirmed_is_refused_in_words(
|
|
client: TestClient, db, registered, served_tree
|
|
):
|
|
"""Rather than the known_hosts prose `connect_kwargs` would raise, which is
|
|
accurate and means nothing to somebody looking at a directory picker."""
|
|
profile = _profile(db, served_tree, host_key="")
|
|
|
|
body = client.get(f"/api/agents/{profile.id}/browse").text
|
|
|
|
assert "has not been confirmed" in body
|
|
|
|
|
|
def test_somebody_elses_connection_is_not_browsable(
|
|
client: TestClient, db, registered, served_tree
|
|
):
|
|
"""These are credentials to somebody's machine, so ownership is the whole
|
|
authorisation -- `sharing.py` grants reading, and a host you can read is a
|
|
host you can log in to.
|
|
|
|
The permission is granted to everybody first, so what is being tested is
|
|
ownership and not the `agent.ssh` gate in front of it. Without that the
|
|
second account is refused before the question is even asked, and the test
|
|
would pass whether or not the ownership check existed.
|
|
"""
|
|
from lembas.services import settings_store
|
|
|
|
profile = _profile(db, served_tree)
|
|
settings_store.update(db, {"default_permissions": {"agent.ssh": True}})
|
|
|
|
client.post("/auth/logout", follow_redirects=False)
|
|
client.post(
|
|
"/auth/register",
|
|
data={"name": "Sam", "email": "sam@shire.test", "password": "potatoes-po-ta-toes"},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
# 404 and not 403: whether that connection exists is not this endpoint's to
|
|
# reveal to somebody who does not own it.
|
|
assert client.get(f"/api/agents/{profile.id}/browse").status_code == 404
|