internal/sshd/sshd.go

e2a32d5f8d59e4213571c602bd9009b6c8fa86ed
gitbay/internal/sshd/sshd.go history · blame · raw

572 lines · 17250 bytes

  1// Package sshd implements the embedded SSH listener: public-key auth against
  2// registered keys, then dispatch to git transport or control commands.
  3package sshd
  4
  5import (
  6	"context"
  7	"crypto/ed25519"
  8	"crypto/rand"
  9	"encoding/base64"
 10	"encoding/pem"
 11	"errors"
 12	"fmt"
 13	"io"
 14	"log/slog"
 15	"maps"
 16	"net"
 17	"os"
 18	"path/filepath"
 19	"slices"
 20	"strconv"
 21	"sync"
 22	"sync/atomic"
 23	"time"
 24
 25	"golang.org/x/crypto/ssh"
 26
 27	"gitbay.org/gitbay/internal/config"
 28	"gitbay.org/gitbay/internal/control"
 29	"gitbay.org/gitbay/internal/gitutil"
 30	"gitbay.org/gitbay/internal/hookd"
 31	"gitbay.org/gitbay/internal/policy"
 32	"gitbay.org/gitbay/internal/protocol"
 33	"gitbay.org/gitbay/internal/store"
 34)
 35
 36type Server struct {
 37	cfg         config.Config
 38	st          *store.Store
 39	sshCfg      *ssh.ServerConfig
 40	authLimiter *rateLimiter
 41	sessions    sync.WaitGroup // accepted connections still being served
 42	mu          sync.Mutex
 43	conns       map[*conn]struct{}
 44	stopping    chan struct{} // closed by Stop
 45	stopOnce    sync.Once
 46}
 47
 48// conn is one accepted connection and how many sessions it is running.
 49// A CLI's shared connection sits idle between commands; on shutdown an
 50// idle connection is closed at once and only a session mid-command is
 51// waited for (#141).
 52type conn struct {
 53	net    net.Conn
 54	active atomic.Int32
 55	// keyID and userID are the key that authenticated the connection and
 56	// its account: 0 before the handshake and for an unregistered key.
 57	// Guarded by Server.mu.
 58	keyID, userID int64
 59	revoked       chan struct{} // closed by cut
 60	cutOnce       sync.Once
 61}
 62
 63// cut ends the connection because its key was revoked: a git transport
 64// on it is killed, and every other command loses its channel.
 65func (c *conn) cut() {
 66	c.cutOnce.Do(func() { close(c.revoked) })
 67	c.net.Close()
 68}
 69
 70func New(cfg config.Config, st *store.Store) (*Server, error) {
 71	s := &Server{cfg: cfg, st: st, authLimiter: newRateLimiter(cfg.Limits.SSHAuthRate, time.Minute), conns: map[*conn]struct{}{}, stopping: make(chan struct{})}
 72
 73	sc := &ssh.ServerConfig{
 74		PublicKeyCallback: s.authenticate,
 75		ServerVersion:     "SSH-2.0-gitbayd",
 76	}
 77	signers, err := loadHostKeys(cfg)
 78	if err != nil {
 79		return nil, err
 80	}
 81	for _, sg := range signers {
 82		sc.AddHostKey(sg)
 83	}
 84	s.sshCfg = sc
 85	st.OnRevoke(s.revoke)
 86	return s, nil
 87}
 88
 89// loadHostKeys loads the configured host keys, or generates an ed25519 key
 90// under server.root/ssh/ when none are configured.
 91func loadHostKeys(cfg config.Config) ([]ssh.Signer, error) {
 92	paths := cfg.SSH.HostKeys
 93	if len(paths) == 0 {
 94		p := filepath.Join(cfg.Server.Root, "ssh", "host_ed25519")
 95		if _, err := os.Stat(p); errors.Is(err, os.ErrNotExist) {
 96			if err := generateHostKey(p); err != nil {
 97				return nil, fmt.Errorf("generating host key: %w", err)
 98			}
 99			slog.Info("generated ssh host key", "path", p)
100		}
101		paths = []string{p}
102	}
103	var signers []ssh.Signer
104	for _, p := range paths {
105		raw, err := os.ReadFile(p)
106		if err != nil {
107			return nil, fmt.Errorf("host key %s: %w", p, err)
108		}
109		sg, err := ssh.ParsePrivateKey(raw)
110		if err != nil {
111			return nil, fmt.Errorf("host key %s: %w", p, err)
112		}
113		signers = append(signers, sg)
114	}
115	return signers, nil
116}
117
118func generateHostKey(path string) error {
119	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
120		return err
121	}
122	_, priv, err := ed25519.GenerateKey(rand.Reader)
123	if err != nil {
124		return err
125	}
126	block, err := ssh.MarshalPrivateKey(priv, "")
127	if err != nil {
128		return err
129	}
130	return os.WriteFile(path, pem.EncodeToMemory(block), 0o600)
131}
132
133// authenticate resolves the presented key to a registered account. The SSH
134// username is ignored; identity comes from the key alone. When registration
135// is open or invite-based, unknown keys are admitted to run exactly one
136// command: register.
137func (s *Server) authenticate(meta ssh.ConnMetadata, pub ssh.PublicKey) (*ssh.Permissions, error) {
138	ip := remoteIP(meta.RemoteAddr())
139	if !s.authLimiter.allow(ip) {
140		// One audit entry per throttled window, not per rejected attempt.
141		if s.authLimiter.firstThrottle(ip) {
142			s.st.Audit(0, "auth.throttled", map[string]any{"ip": ip, "rate": s.cfg.Limits.SSHAuthRate})
143		}
144		return nil, fmt.Errorf("too many authentication attempts; try again shortly")
145	}
146	fp := ssh.FingerprintSHA256(pub)
147	key, err := s.st.SSHKeyByFingerprint(fp)
148	if err != nil && !errors.Is(err, store.ErrNotFound) {
149		// The store, not the key, failed. Neither a failure against the
150		// limiter nor "unknown key": a busy database during a restart
151		// would otherwise lock every client out for a minute.
152		slog.Error("ssh auth: key lookup", "err", err)
153		return nil, fmt.Errorf("authentication temporarily unavailable")
154	}
155	if err != nil {
156		if s.cfg.Registration.Mode != "closed" {
157			return &ssh.Permissions{Extensions: map[string]string{
158				"anon-key": base64.StdEncoding.EncodeToString(pub.Marshal()),
159			}}, nil
160		}
161		s.authLimiter.fail(ip)
162		s.st.Audit(0, "auth.failed", map[string]any{"ip": ip, "fingerprint": fp})
163		return nil, fmt.Errorf("unknown key %s", fp)
164	}
165	s.authLimiter.success(ip)
166	return &ssh.Permissions{Extensions: map[string]string{
167		"user-id": strconv.FormatInt(key.UserID, 10),
168		"key-id":  strconv.FormatInt(key.ID, 10),
169	}}, nil
170}
171
172// Serve accepts connections on ln until it is closed.
173func (s *Server) Serve(ln net.Listener) error {
174	served := make(chan struct{})
175	defer close(served)
176	go s.sweep(served)
177	for {
178		nc, err := ln.Accept()
179		if err != nil {
180			return err
181		}
182		c := &conn{net: nc, revoked: make(chan struct{})}
183		s.mu.Lock()
184		s.conns[c] = struct{}{}
185		s.mu.Unlock()
186		s.sessions.Add(1)
187		go func() {
188			defer s.sessions.Done()
189			defer func() {
190				s.mu.Lock()
191				delete(s.conns, c)
192				s.mu.Unlock()
193			}()
194			s.handleConn(c)
195		}()
196	}
197}
198
199// revoke closes the connections opened by the keys r names.
200func (s *Server) revoke(r store.Revoked) {
201	var cut []*conn
202	s.mu.Lock()
203	for c := range s.conns {
204		if c.keyID == 0 {
205			continue
206		}
207		if (r.UserID != 0 && c.userID == r.UserID) || slices.Contains(r.KeyIDs, c.keyID) {
208			cut = append(cut, c)
209		}
210	}
211	s.mu.Unlock()
212	for _, c := range cut {
213		c.cut()
214	}
215}
216
217// sweepInterval bounds how long a revocation this process was not told
218// about (gitbayd admin on the host) leaves a connection open.
219const sweepInterval = 15 * time.Second
220
221func (s *Server) sweep(served <-chan struct{}) {
222	t := time.NewTicker(sweepInterval)
223	defer t.Stop()
224	for {
225		select {
226		case <-t.C:
227			s.sweepOnce()
228		case <-served:
229			return
230		case <-s.stopping:
231			return
232		}
233	}
234}
235
236// sweepOnce cuts every connection whose key is no longer live. Only
237// connections whose key was asked about are judged: one that
238// authenticated while the query ran waits for the next sweep.
239func (s *Server) sweepOnce() {
240	asked := map[int64]bool{}
241	s.mu.Lock()
242	for c := range s.conns {
243		if c.keyID != 0 {
244			asked[c.keyID] = true
245		}
246	}
247	s.mu.Unlock()
248	if len(asked) == 0 {
249		return
250	}
251	live, err := s.st.LiveSSHKeys(slices.Collect(maps.Keys(asked)))
252	if err != nil {
253		slog.Error("ssh sweep: key lookup", "err", err)
254		return
255	}
256	var cut []*conn
257	s.mu.Lock()
258	for c := range s.conns {
259		if asked[c.keyID] && !live[c.keyID] {
260			cut = append(cut, c)
261		}
262	}
263	s.mu.Unlock()
264	for _, c := range cut {
265		c.cut()
266	}
267}
268
269// Stop ends the commands that run until something happens (build log
270// --follow), so a shutdown drain waits only for work that finishes. It
271// does not close connections; Shutdown does.
272func (s *Server) Stop() {
273	s.stopOnce.Do(func() { close(s.stopping) })
274}
275
276// Shutdown closes every idle connection, then waits for the ones with a
277// session running, or for ctx. The caller closes the listener first; a
278// push in flight completes rather than being cut mid-pack.
279func (s *Server) Shutdown(ctx context.Context) error {
280	s.Stop()
281	s.mu.Lock()
282	for c := range s.conns {
283		if c.active.Load() == 0 {
284			c.net.Close()
285		}
286	}
287	s.mu.Unlock()
288	done := make(chan struct{})
289	go func() {
290		s.sessions.Wait()
291		close(done)
292	}()
293	select {
294	case <-done:
295		return nil
296	case <-ctx.Done():
297		return ctx.Err()
298	}
299}
300
301func (s *Server) handleConn(c *conn) {
302	defer c.net.Close()
303	sconn, chans, reqs, err := ssh.NewServerConn(c.net, s.sshCfg)
304	if err != nil {
305		return
306	}
307	defer sconn.Close()
308	ext := sconn.Permissions.Extensions
309	s.mu.Lock()
310	c.keyID, _ = strconv.ParseInt(ext["key-id"], 10, 64)
311	c.userID, _ = strconv.ParseInt(ext["user-id"], 10, 64)
312	s.mu.Unlock()
313	go ssh.DiscardRequests(reqs)
314
315	for newCh := range chans {
316		if newCh.ChannelType() != "session" {
317			newCh.Reject(ssh.UnknownChannelType, "only session channels are supported")
318			continue
319		}
320		ch, chReqs, err := newCh.Accept()
321		if err != nil {
322			continue
323		}
324		c.active.Add(1)
325		go func() {
326			defer c.active.Add(-1)
327			s.handleSession(c, sconn, ch, chReqs)
328		}()
329	}
330}
331
332func (s *Server) handleSession(c *conn, sconn *ssh.ServerConn, ch ssh.Channel, reqs <-chan *ssh.Request) {
333	defer ch.Close()
334	var term control.Term
335	for req := range reqs {
336		switch req.Type {
337		case "exec":
338			var payload struct{ Command string }
339			if err := ssh.Unmarshal(req.Payload, &payload); err != nil {
340				req.Reply(false, nil)
341				continue
342			}
343			req.Reply(true, nil)
344			// x/crypto closes reqs when the client closes the channel. That
345			// is how a follow learns nobody is reading: the CLI's shared
346			// connection outlives a Ctrl-C, the channel does not. Stop
347			// ends it too, for a restart.
348			closed := make(chan struct{})
349			go func() {
350				for r := range reqs {
351					r.Reply(false, nil)
352				}
353				close(closed)
354			}()
355			done := make(chan struct{})
356			go func() {
357				select {
358				case <-closed:
359				case <-s.stopping:
360				}
361				close(done)
362			}()
363			code := s.runExec(c, sconn, ch, term, payload.Command, done)
364			sendExit(ch, code)
365			return
366		case "shell":
367			req.Reply(true, nil)
368			fmt.Fprintf(ch, "gitbay control plane: interactive shells are not available.\nTry: ssh %s help\n", s.cfg.Server.SiteURL)
369			sendExit(ch, protocol.ExitUsage)
370			return
371		case "env":
372			var kv struct{ Name, Value string }
373			if ssh.Unmarshal(req.Payload, &kv) == nil && kv.Name == "GITBAY_TERM" {
374				term = control.ParseTerm(kv.Value)
375			}
376			req.Reply(true, nil)
377		case "pty-req":
378			// Harmless; accept and ignore.
379			req.Reply(true, nil)
380		default:
381			req.Reply(false, nil)
382		}
383	}
384}
385
386func sendExit(ch ssh.Channel, code int) {
387	var msg = struct{ Status uint32 }{uint32(code)}
388	ch.SendRequest("exit-status", false, ssh.Marshal(&msg))
389}
390
391func (s *Server) runExec(c *conn, sconn *ssh.ServerConn, ch ssh.Channel, term control.Term, cmdline string, done <-chan struct{}) int {
392	ext := sconn.Permissions.Extensions
393	if blob := ext["anon-key"]; blob != "" {
394		return s.runAnonymous(ch, blob, cmdline)
395	}
396	userID, _ := strconv.ParseInt(ext["user-id"], 10, 64)
397	keyID, _ := strconv.ParseInt(ext["key-id"], 10, 64)
398	// A connection outlives its commands, so the key is read again for
399	// each one: what it may do is what it may do now (#256).
400	key, err := s.st.SSHKeyByID(keyID)
401	if errors.Is(err, store.ErrNotFound) || (err == nil && key.UserID != userID) {
402		fmt.Fprintln(ch.Stderr(), "this key is no longer registered")
403		return protocol.ExitDenied
404	}
405	if err != nil {
406		slog.Error("ssh exec: key lookup", "err", err)
407		fmt.Fprintln(ch.Stderr(), "authentication temporarily unavailable")
408		return protocol.ExitFailure
409	}
410	user, err := s.st.UserByID(userID)
411	if err != nil {
412		fmt.Fprintln(ch.Stderr(), "account no longer exists")
413		return protocol.ExitDenied
414	}
415	_ = s.st.TouchSSHKey(keyID)
416	return Exec(s.cfg, s.st, user, key, term, cmdline, ch, ch, ch.Stderr(), done, s.stopping, c.revoked)
417}
418
419// runAnonymous handles a session from an unregistered key: the register
420// command and nothing else.
421func (s *Server) runAnonymous(ch ssh.Channel, keyB64, cmdline string) int {
422	raw, err := base64.StdEncoding.DecodeString(keyB64)
423	if err != nil {
424		return protocol.ExitFailure
425	}
426	pub, err := ssh.ParsePublicKey(raw)
427	if err != nil {
428		return protocol.ExitFailure
429	}
430	argv, err := protocol.Tokenize(cmdline)
431	if err != nil {
432		fmt.Fprintf(ch.Stderr(), "cannot parse command: %v\n", err)
433		return protocol.ExitUsage
434	}
435	if len(argv) == 0 || argv[0] != "register" {
436		fmt.Fprintf(ch.Stderr(), "this key is not registered here. Create an account with:\n  ssh <host> register --username <name> %s\n",
437			map[string]string{"open": "--email <address>", "invite": "--invite <code>"}[s.cfg.Registration.Mode])
438		return protocol.ExitDenied
439	}
440	return control.RunRegister(s.cfg, s.st, pub, argv, ch, ch.Stderr())
441}
442
443// Exec runs one SSH exec command line for an authenticated key. It is the
444// single dispatch path shared by the embedded listener and the system-sshd
445// forced command (gitbayd shell). Closing revoked kills a git transport.
446func Exec(cfg config.Config, st *store.Store, user store.User, key store.SSHKey, term control.Term, cmdline string,
447	stdin io.Reader, stdout, stderr io.Writer, done, stopping, revoked <-chan struct{}) int {
448	if user.Disabled {
449		fmt.Fprintln(stderr, "this account is disabled; contact the instance admin")
450		return protocol.ExitDenied
451	}
452	argv, err := protocol.Tokenize(cmdline)
453	if err != nil {
454		fmt.Fprintf(stderr, "cannot parse command: %v\n", err)
455		return protocol.ExitUsage
456	}
457	if len(argv) > 0 {
458		switch argv[0] {
459		case "git-upload-pack", "git-receive-pack", "git-upload-archive":
460			if user.Pending {
461				fmt.Fprintln(stderr, "your account is not active yet: verify your email first")
462				return protocol.ExitDenied
463			}
464			return runGit(cfg, st, user, key.Scope, argv, stdin, stdout, stderr, revoked)
465		case "git-lfs-authenticate":
466			// Part of the git transport, not the control plane: usable by
467			// git-scoped and deploy keys, with the transports' access rules.
468			if user.Pending {
469				fmt.Fprintln(stderr, "your account is not active yet: verify your email first")
470				return protocol.ExitDenied
471			}
472			return runLFSAuthenticate(cfg, st, user, key.Scope, argv, stdout, stderr)
473		}
474	}
475	ctx := &control.Ctx{
476		User:     user,
477		Scope:    key.Scope,
478		Source:   key.Fingerprint,
479		Term:     term,
480		Store:    st,
481		Cfg:      cfg,
482		Stdin:    stdin,
483		Stdout:   stdout,
484		Stderr:   stderr,
485		Done:     done,
486		Stopping: stopping,
487	}
488	return control.Dispatch(ctx, argv)
489}
490
491// runGit streams a git transport service after access checks.
492func runGit(cfg config.Config, st *store.Store, user store.User, scope string, argv []string,
493	stdin io.Reader, stdout, stderr io.Writer, revoked <-chan struct{}) int {
494	service := argv[0]
495	if len(argv) != 2 {
496		fmt.Fprintf(stderr, "usage: %s <path>\n", service)
497		return protocol.ExitUsage
498	}
499	write := service == "git-receive-pack"
500
501	repo, err := st.RepoByPath(argv[1])
502	if err != nil {
503		fmt.Fprintln(stderr, "repository not found")
504		return protocol.ExitNotFound
505	}
506	if policy.IsDeployScope(scope) {
507		// A deploy key authorizes by its binding alone: one repository,
508		// its mode, nothing inherited from whoever registered it. Any
509		// mismatch reads as nonexistence, same as the access rules.
510		if !policy.DeployScopeAllows(scope, repo.ID, write) {
511			fmt.Fprintln(stderr, "repository not found")
512			return protocol.ExitNotFound
513		}
514	} else {
515		grant, err := st.AccessRole(repo.ID, user.ID)
516		if err != nil {
517			fmt.Fprintln(stderr, "internal error")
518			return protocol.ExitFailure
519		}
520		if !policy.CanRead(user, repo, grant) {
521			// Same answer as nonexistence: private repos must not be enumerable.
522			fmt.Fprintln(stderr, "repository not found")
523			return protocol.ExitNotFound
524		}
525		if !policy.ScopeAllowsGit(scope, repo.Path(), write) {
526			fmt.Fprintf(stderr, "this key's scope (%s) does not allow %s on %s\n", scope, service, repo.Path())
527			return protocol.ExitDenied
528		}
529		if write && !policy.CanWrite(user, repo, grant) {
530			fmt.Fprintf(stderr, "write access to %s denied\n", repo.Path())
531			return protocol.ExitDenied
532		}
533	}
534	if write && repo.Settings.Archived {
535		fmt.Fprintf(stderr, "%s is archived and read-only\n", repo.Path())
536		return protocol.ExitDenied
537	}
538	if write {
539		if mirrored, err := st.PullMirrored(repo.ID); err == nil && mirrored {
540			fmt.Fprintf(stderr, "%s is a pull mirror: its refs come from the upstream; push there instead\n", repo.Path())
541			return protocol.ExitDenied
542		}
543	}
544
545	dir := control.RepoDir(cfg.Server.Root, repo.OwnerName, repo.Name)
546	env := []string{
547		hookd.EnvSocket + "=" + hookd.SocketPath(cfg.Server.Root),
548		hookd.EnvRepoID + "=" + strconv.FormatInt(repo.ID, 10),
549		hookd.EnvUserID + "=" + strconv.FormatInt(user.ID, 10),
550		hookd.EnvScope + "=" + scope,
551	}
552	// A storage quota on the owner rides the same mechanism as the pack
553	// cap: the pack may be no larger than what the owner has left.
554	maxPack := cfg.Limits.MaxPackBytes
555	if write && repo.OwnerKind == "user" {
556		if limit := control.ByteLimit(st, control.QuotaConfig(cfg), repo.OwnerID); limit > 0 {
557			used := control.OwnedBytes(st, cfg.Server.Root, repo.OwnerID)
558			left := limit - used
559			if left <= 0 {
560				fmt.Fprintf(stderr, "%s's storage quota is used up (%d of %d bytes); delete something, or ask an admin to raise the limit\n", repo.OwnerName, used, limit)
561				return protocol.ExitDenied
562			}
563			if maxPack == 0 || left < maxPack {
564				maxPack = left
565			}
566		}
567	}
568	if err := gitutil.Transport(service, dir, stdin, stdout, stderr, env, maxPack, revoked); err != nil {
569		return protocol.ExitFailure
570	}
571	return protocol.ExitOK
572}