SSH connections, kept by the people who own them

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 c3d6660881
commit fe7227af62
21 changed files with 2452 additions and 7 deletions
+103
View File
@@ -0,0 +1,103 @@
"""SSH connections an agent chat can act through.
User-owned, like a `Note` and unlike a `Connection`. That is the opposite of
the rule custom tools and MCP servers follow, and the difference is the point:
those are instance configuration an administrator could grant themselves in one
click anyway, while this is somebody's own machine and somebody's own key.
"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.
**Nothing an agent does runs on the LLeMbas machine.** A local sandbox was
designed and dropped: every hard problem in it came from executing on the host
that holds the database and the encryption key. Over SSH, isolation is whatever
host somebody points this at -- which means the security of an agent chat is the
security of that host, and nothing here can tell a throwaway container from a
production server. The admin copy says so out loud.
"""
from __future__ import annotations
from datetime import datetime
from typing import TYPE_CHECKING, Any
from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, Text, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column, relationship
from lembas.db.base import Base, Timestamps, UUIDPrimaryKey
from lembas.db.types import JSONDict
if TYPE_CHECKING: # pragma: no cover - annotation only
from lembas.db.models.user import User
# How the connection authenticates.
AUTH_KEY = "key"
AUTH_PASSWORD = "password"
AUTH_METHODS = (AUTH_KEY, AUTH_PASSWORD)
class SshProfile(UUIDPrimaryKey, Timestamps, Base):
"""One host somebody can point an agent chat at."""
__tablename__ = "ssh_profiles"
__table_args__ = (UniqueConstraint("owner_id", "name", name="uq_ssh_profile_name"),)
owner_id: Mapped[str] = mapped_column(
String(32), ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
name: Mapped[str] = mapped_column(String(120), nullable=False)
host: Mapped[str] = mapped_column(String(255), nullable=False)
port: Mapped[int] = mapped_column(Integer, default=22, nullable=False)
username: Mapped[str] = mapped_column(String(120), nullable=False)
auth: Mapped[str] = mapped_column(String(16), default=AUTH_KEY, nullable=False)
password_encrypted: Mapped[str] = mapped_column(Text, default="")
private_key_encrypted: Mapped[str] = mapped_column(Text, default="")
key_passphrase_encrypted: Mapped[str] = mapped_column(Text, default="")
# One OpenSSH known_hosts line, captured the first time this host answered
# and shown as a fingerprint to be confirmed, then pinned. Empty means
# "never seen". Handed to asyncssh as `known_hosts=<these bytes>` and never
# as None, which turns host key checking off altogether.
host_key: Mapped[str] = mapped_column(Text, default="")
# The SHA256 fingerprint of the above, so the profile page can show what was
# accepted without parsing the line again on every render.
host_fingerprint: Mapped[str] = mapped_column(String(120), default="")
# Where a chat starts by default. A chat records its own, chosen when it is
# created and fixed thereafter; this is only the suggestion in the picker.
default_dir: Mapped[str] = mapped_column(String(500), default="")
connect_timeout: Mapped[int] = mapped_column(Integer, default=15, nullable=False)
enabled: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
# What the last connection attempt found, for the list. `server_banner` is
# whatever the host said about itself -- useful for telling two containers
# apart.
last_checked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
last_error: Mapped[str] = mapped_column(Text, default="")
server_info: Mapped[dict[str, Any]] = mapped_column(JSONDict, default=dict)
owner: Mapped[User] = relationship()
@property
def label(self) -> str:
return self.name or f"{self.username}@{self.host}"
@property
def address(self) -> str:
return f"{self.username}@{self.host}" + (f":{self.port}" if self.port != 22 else "")
@property
def verified(self) -> bool:
"""Whether this host's key has been seen and pinned."""
return bool(self.host_key)
def __repr__(self) -> str:
return f"<SshProfile {self.name} {self.address}>"
__all__ = ["AUTH_KEY", "AUTH_METHODS", "AUTH_PASSWORD", "SshProfile"]