"""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=` 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"" __all__ = ["AUTH_KEY", "AUTH_METHODS", "AUTH_PASSWORD", "SshProfile"]