""" Model for terminal sessions - stores SSH terminal session metadata. """ from __future__ import annotations from datetime import datetime from typing import Optional from sqlalchemy import DateTime, Index, Integer, String, text from sqlalchemy.orm import Mapped, mapped_column from sqlalchemy.sql import func from .database import Base # Session status constants SESSION_STATUS_STARTING = "starting" SESSION_STATUS_ACTIVE = "active" SESSION_STATUS_CLOSING = "closing" SESSION_STATUS_CLOSED = "closed" SESSION_STATUS_EXPIRED = "expired" SESSION_STATUS_ERROR = "error" # Close reason constants CLOSE_REASON_USER = "user_close" CLOSE_REASON_TTL = "ttl" CLOSE_REASON_IDLE = "idle" CLOSE_REASON_QUOTA = "quota" CLOSE_REASON_SERVER_SHUTDOWN = "server_shutdown" CLOSE_REASON_CLIENT_LOST = "client_lost" CLOSE_REASON_REUSED = "reused" class TerminalSession(Base): """ Represents an active or recent SSH terminal session. Sessions are created when a user opens a terminal to a host, and cleaned up after expiration or manual closure. Status lifecycle: - starting: Session being created, ttyd spawning - active: Session running, client connected - closing: Session being terminated - closed: Session terminated normally - expired: Session terminated due to TTL/idle timeout - error: Session terminated due to error """ __tablename__ = "terminal_sessions" __table_args__ = ( Index('ix_terminal_sessions_user_status', 'user_id', 'status'), Index('ix_terminal_sessions_last_seen', 'last_seen_at'), ) id: Mapped[str] = mapped_column(String(64), primary_key=True) host_id: Mapped[str] = mapped_column(String, nullable=False, index=True) host_name: Mapped[str] = mapped_column(String, nullable=False) host_ip: Mapped[str] = mapped_column(String, nullable=False) user_id: Mapped[Optional[str]] = mapped_column(String, nullable=True) username: Mapped[Optional[str]] = mapped_column(String, nullable=True) # Token hash for session authentication (never store plain token) token_hash: Mapped[str] = mapped_column(String(128), nullable=False) # ttyd process management ttyd_port: Mapped[int] = mapped_column(Integer, nullable=False) ttyd_pid: Mapped[Optional[int]] = mapped_column(Integer, nullable=True) # Session mode: 'embedded' or 'popout' mode: Mapped[str] = mapped_column(String(20), nullable=False, server_default=text("'embedded'")) # Session status: 'starting', 'active', 'closing', 'closed', 'expired', 'error' status: Mapped[str] = mapped_column(String(20), nullable=False, server_default=text("'active'")) # Close reason: 'user_close', 'ttl', 'idle', 'quota', 'server_shutdown', 'client_lost', 'reused' reason_closed: Mapped[Optional[str]] = mapped_column(String(30), nullable=True) # Timestamps created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) last_seen_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) closed_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True) def __repr__(self) -> str: return f""