1package serve23import (4 "context"5 "errors"6 "fmt"7 "io"8 "log/slog"9 "net"10 "sync"11 "sync/atomic"12 "time"1314 tea "charm.land/bubbletea/v2"15 "charm.land/ssh"16 bm "charm.land/wish/v2/bubbletea"17 gossh "golang.org/x/crypto/ssh"1819 "src.rest/cat/012/internal/confine"20 "src.rest/cat/012/internal/jev"21 "src.rest/cat/012/internal/room"22 "src.rest/cat/012/internal/sheet"23 "src.rest/cat/012/internal/telemetry"24 "src.rest/cat/012/internal/ui"25)2627// handshakeTimeout bounds the time from connecting to logging in.28const handshakeTimeout = 30 * time.Second2930// Server serves 012 sessions over SSH.31type Server struct {32 opts Options33 root confine.Root34 log *slog.Logger35 jev jev.Client // nil: JEV functions are off36 ssh *ssh.Server37 hostPub gossh.PublicKey38 slots chan struct{} // one per running session39 active atomic.Int6440 rooms *room.Registry // the shared workbooks; nil with sharing off4142 // stop is closed when the server shuts down, ending every session,43 // which keeps its unsaved work first; running counts the sessions44 // still doing so. mu orders sessions starting against stopping.45 mu sync.Mutex46 stopped bool47 stop chan struct{}48 running sync.WaitGroup49}5051// halt tells every session to end; no new one starts after it.52func (s *Server) halt() {53 s.mu.Lock()54 defer s.mu.Unlock()55 if !s.stopped {56 s.stopped = true57 close(s.stop)58 }59}6061// begin counts a session as running, or reports false when the server62// is stopping. A session that began calls s.running.Done when it ends.63func (s *Server) begin() bool {64 s.mu.Lock()65 defer s.mu.Unlock()66 if s.stopped {67 return false68 }69 s.running.Add(1)70 return true71}7273// Deps are what a Server takes from its surroundings rather than its74// options.75type Deps struct {76 // Log gets a line per login, rejection and session, besides the77 // telemetry events; nil discards them.78 Log *slog.Logger79 // JEV answers JEV functions, with a cache per session; nil turns them80 // off.81 JEV jev.Client82}8384// New checks o and prepares a server: the served directory, the host85// key (generated if missing) and the authorized keys must be usable.86func New(o Options, d Deps) (*Server, error) {87 if err := o.validate(); err != nil {88 return nil, err89 }90 root, err := confine.New(o.Dir)91 if err != nil {92 return nil, fmt.Errorf("served directory: %w", err)93 }94 keys, _, err := authorizedKeys(o.AuthorizedKeys)95 if err != nil {96 return nil, fmt.Errorf("authorized keys: %w", err)97 }98 if len(keys) == 0 {99 return nil, fmt.Errorf("authorized keys: no usable keys in %s", o.AuthorizedKeys)100 }101 signer, err := hostKey(o.HostKey)102 if err != nil {103 return nil, err104 }105 if d.Log == nil {106 d.Log = slog.New(slog.DiscardHandler)107 }108 s := &Server{opts: o, root: root, log: d.Log, jev: d.JEV, hostPub: signer.PublicKey(), slots: make(chan struct{}, o.MaxSessions),109 stop: make(chan struct{})}110 switch o.Share {111 case "edit":112 s.rooms = room.NewRegistry(room.Edit)113 case "view":114 s.rooms = room.NewRegistry(room.View)115 }116 s.ssh = &ssh.Server{117 Addr: o.Listen,118 Handler: s.session,119 HostSigners: []ssh.Signer{signer},120 Version: "012",121 PublicKeyHandler: s.authorize,122 // No password or keyboard-interactive auth, no forwarding of any123 // kind, no global requests and no subsystems: only sessions.124 ChannelHandlers: map[string]ssh.ChannelHandler{"session": ssh.DefaultSessionHandler},125 RequestHandlers: map[string]ssh.RequestHandler{},126 SubsystemHandlers: map[string]ssh.SubsystemHandler{},127 SessionRequestCallback: allowSession,128 ConnectionFailedCallback: func(c net.Conn, err error) {129 s.event("ssh.failed", slog.String("remote", c.RemoteAddr().String()), slog.String("error", err.Error()))130 },131 HandshakeTimeout: handshakeTimeout,132 }133 if o.IdleTimeout > 0 {134 // A backstop for connections without a session. Sessions end135 // on their own after IdleTimeout without input (see run), which136 // keepalives don't reset.137 s.ssh.IdleTimeout = o.IdleTimeout + time.Minute138 }139 return s, nil140}141142// allowSession lets a session start 012 and nothing else. A shell143// request starts it on a new sheet. An exec request is taken only as the144// name of a file to open, with a terminal and in one word (ssh -t host145// file.012); it is never run. Other exec requests (ssh host command) and146// subsystems (sftp) are refused.147func allowSession(sess ssh.Session, requestType string) bool {148 switch requestType {149 case "shell":150 return true151 case "exec":152 _, _, pty := sess.Pty()153 return pty && len(sess.Command()) == 1154 }155 return false156}157158// HostKey is the server's public key, for known_hosts.159func (s *Server) HostKey() gossh.PublicKey { return s.hostPub }160161// Dir is the served directory, with links resolved.162func (s *Server) Dir() string { return s.root.Dir() }163164// Sessions is how many sessions are running.165func (s *Server) Sessions() int { return int(s.active.Load()) }166167// ListenAndServe listens on the options' address and serves until168// Shutdown or Close.169func (s *Server) ListenAndServe() error {170 l, err := net.Listen("tcp", s.opts.Listen)171 if err != nil {172 return err173 }174 return s.Serve(l)175}176177// Serve serves sessions arriving on l.178func (s *Server) Serve(l net.Listener) error {179 s.log.Info("serving", "addr", l.Addr().String(), "dir", s.root.Dir(), "host_key", fingerprint(s.hostPub))180 err := s.ssh.Serve(l)181 if errors.Is(err, ssh.ErrServerClosed) {182 return nil183 }184 return err185}186187// Shutdown ends every session, which keeps its unsaved work in a188// recovery file and tells the client, waits for them to end until ctx is189// done, then closes the listener and the connections left. Those hold no190// session, and a connection can hold nothing else.191func (s *Server) Shutdown(ctx context.Context) error {192 s.halt()193 err := s.waitSessions(ctx)194 if cerr := s.ssh.Close(); err == nil {195 err = cerr196 }197 return err198}199200// closeWait bounds how long Close waits for sessions to keep their201// unsaved work.202const closeWait = 5 * time.Second203204// Close stops the server and drops every connection at once. Sessions205// still keep their unsaved work, for up to closeWait.206func (s *Server) Close() error {207 s.halt()208 err := s.ssh.Close()209 ctx, cancel := context.WithTimeout(context.Background(), closeWait)210 defer cancel()211 s.waitSessions(ctx)212 return err213}214215// waitSessions waits for running sessions to end, until ctx is done.216func (s *Server) waitSessions(ctx context.Context) error {217 done := make(chan struct{})218 go func() {219 s.running.Wait()220 close(done)221 }()222 select {223 case <-done:224 return nil225 case <-ctx.Done():226 return ctx.Err()227 }228}229230// authorize accepts key if the authorized_keys file lists it, reading231// the file again so edits apply without a restart.232func (s *Server) authorize(ctx ssh.Context, key ssh.PublicKey) bool {233 keys, _, err := authorizedKeys(s.opts.AuthorizedKeys)234 if err != nil {235 s.event("ssh.auth_error", slog.String("error", err.Error()))236 return false237 }238 for _, k := range keys {239 if ssh.KeysEqual(key, k) {240 return true241 }242 }243 s.event("ssh.rejected", slog.String("user", ctx.User()), slog.String("key", fingerprint(key)),244 slog.String("remote", ctx.RemoteAddr().String()))245 return false246}247248// event logs name to telemetry and the server log. Events name users,249// keys and addresses, never what's in files.250func (s *Server) event(name string, attrs ...slog.Attr) {251 telemetry.Event(name, 0, attrs...)252 s.log.LogAttrs(context.Background(), slog.LevelInfo, name, attrs...)253}254255// session runs one 012 over one SSH session.256func (s *Server) session(sess ssh.Session) {257 who := []slog.Attr{slog.String("user", sess.User()), slog.String("key", fingerprint(sess.PublicKey())),258 slog.String("remote", sess.RemoteAddr().String())}259 pty, winch, ok := sess.Pty()260 if !ok {261 fmt.Fprint(sess.Stderr(), "012 needs a terminal: connect with ssh -t\r\n")262 sess.Exit(1)263 return264 }265 name, err := s.fileArg(sess)266 if err != nil {267 fmt.Fprintf(sess.Stderr(), "012: %v\r\n", err)268 sess.Exit(1)269 return270 }271 if !s.begin() {272 fmt.Fprint(sess.Stderr(), "012: the server is stopping\r\n")273 sess.Exit(1)274 return275 }276 defer s.running.Done()277 select {278 case s.slots <- struct{}{}:279 default:280 s.event("ssh.full", who...)281 fmt.Fprintf(sess.Stderr(), "012: %d sessions are running, the most this server allows; try again later\r\n", s.opts.MaxSessions)282 sess.Exit(1)283 return284 }285 n := s.active.Add(1)286 telemetry.Set("ssh_sessions", n)287 s.event("ssh.session", append(who, slog.String("term", pty.Term), slog.Int("width", pty.Window.Width), slog.Int("height", pty.Window.Height),288 slog.Bool("file", name != ""))...)289 start := time.Now()290 defer func() {291 telemetry.Set("ssh_sessions", s.active.Add(-1))292 <-s.slots293 }()294295 // The session's span holds what its program does, in one trace.296 span := telemetry.Start("serve.session", slog.String("term", pty.Term))297 end := s.run(sess, pty, winch, name, span.Parent())298 span.End(slog.Bool("idle", end.idle))299 d := time.Since(start)300 s.event("ssh.session_end", append(who, slog.Duration("duration", d), slog.Bool("idle", end.idle),301 slog.Bool("stopped", end.stopped), slog.Bool("recovered", end.kept != ""))...)302 end.tell(sess, s.opts.IdleTimeout, s.root)303 sess.Exit(0)304}305306// run runs the program, on the file name when there's one, until it307// quits, the client goes away, the session goes idle or the server308// stops, and keeps unsaved work for the last two.309func (s *Server) run(sess ssh.Session, pty ssh.Pty, winch <-chan ssh.Window, name string, parent telemetry.Parent) ending {310 m := ui.New(sheet.New(), "")311 m.TraceUnder(parent)312 env := append(sess.Environ(), "TERM="+pty.Term)313 m.Serve(s.root, env)314 if s.rooms != nil {315 m.ShareRooms(s.rooms, sess.User())316 }317 m.OpenOnStart(name)318 if s.jev != nil {319 // Each session has its own cache: it queues that session's320 // questions, and its answers recalculate that session's sheets.321 m.EnableJEV(s.jev, jev.NewCache())322 }323 in := newActivity(sess)324 g := ui.Guard(m)325 var model tea.Model = g326 var shared *ui.Shared327 if s.rooms != nil {328 shared = ui.InRooms(g)329 model = shared330 }331 p := tea.NewProgram(model, append(bm.MakeOptions(sess), tea.WithInput(in), tea.WithFPS(ui.FrameRate))...)332333 // The connection's context ends when the client goes away; cancel334 // stops the goroutines below when the program quits first.335 ctx, cancel := context.WithCancel(sess.Context())336 defer cancel()337 if shared != nil {338 shared.Attach(ctx, p)339 }340 go resize(ctx, p, winch)341 go func() {342 select {343 case <-s.stop:344 p.Quit()345 case <-ctx.Done():346 }347 }()348 var idled atomic.Bool349 if s.opts.IdleTimeout > 0 {350 go func() {351 if in.waitIdle(ctx, s.opts.IdleTimeout) {352 idled.Store(true)353 p.Quit()354 }355 }()356 }357 _, err := p.Run()358 if err != nil && !errors.Is(err, tea.ErrProgramKilled) {359 s.log.Error("session", "error", err.Error())360 }361 p.Kill()362 end := ending{idle: idled.Load(), stopped: s.stopping(), crash: g.Finish(err)}363 if c := end.crash; c != nil {364 // The report is the server's log: the operator's to read.365 s.log.Error("session panic", "where", c.Where, "panic", fmt.Sprint(c.Value), "stack", string(c.Stack))366 }367 // A shared workbook stays with those still in its room; the last368 // one out keeps its unsaved work.369 _, end.others = g.Model().Others()370 if g.Model().LeaveRoom() {371 end.others = ""372 end.keep(g)373 }374 return end375}376377// stopping reports whether the server is shutting down.378func (s *Server) stopping() bool {379 select {380 case <-s.stop:381 return true382 default:383 return false384 }385}386387// resize tells p about the client's window changes, and quits it when388// ctx ends.389func resize(ctx context.Context, p *tea.Program, winch <-chan ssh.Window) {390 for {391 select {392 case <-ctx.Done():393 p.Quit()394 return395 case w, ok := <-winch:396 if !ok {397 <-ctx.Done()398 p.Quit()399 return400 }401 p.Send(tea.WindowSizeMsg{Width: w.Width, Height: w.Height})402 }403 }404}405406// activity is a session's input, noting when it last had any.407type activity struct {408 r io.Reader409 last atomic.Int64 // UnixNano of the last read that returned bytes410}411412func newActivity(r io.Reader) *activity {413 a := &activity{r: r}414 a.last.Store(time.Now().UnixNano())415 return a416}417418func (a *activity) Read(p []byte) (int, error) {419 n, err := a.r.Read(p)420 if n > 0 {421 a.last.Store(time.Now().UnixNano())422 }423 return n, err424}425426// waitIdle waits until there's been no input for d, reporting true, or427// until ctx ends, reporting false.428func (a *activity) waitIdle(ctx context.Context, d time.Duration) bool {429 t := time.NewTimer(d)430 defer t.Stop()431 for {432 select {433 case <-ctx.Done():434 return false435 case now := <-t.C:436 left := d - now.Sub(time.Unix(0, a.last.Load()))437 if left <= 0 {438 return true439 }440 t.Reset(left)441 }442 }443}