Commit `c0969481b0`

c0969481b067872df263c8f577ac712cefc7b682

parent: bf64c30c08

Verified · cmc

cmc <hello@cleberg.net> · 2026-09-28 06:36 UTC

sshd: re-read the key per exec; revocation cuts its connections

Ref #256

Layout: unified · split

cmd/gitbayd/system.go +1 −1

		@@ -94,7 +94,7 @@ func shellCmd() *cobra.Command {
94	94	fmt.Fprintf(os.Stderr, "gitbay control plane: interactive shells are not available.\nTry: ssh <host> help\n")
95	95	os.Exit(protocol.ExitUsage)
96	96	}
97		code := sshd.Exec(cfg, st, user, key.Scope, key.Fingerprint, control.ParseTerm(os.Getenv("GITBAY_TERM")), cmdline, os.Stdin, os.Stdout, os.Stderr, nil, nil)
	97	code := sshd.Exec(cfg, st, user, key, control.ParseTerm(os.Getenv("GITBAY_TERM")), cmdline, os.Stdin, os.Stdout, os.Stderr, nil, nil, nil)
98	98	st.Close()
99	99	os.Exit(code)
100	100	return nil

internal/sshd/revoke_test.go added +112

		@@ -0,0 +1,112 @@
	1	package sshd
	2
	3	import (
	4	"bytes"
	5	"errors"
	6	"strings"
	7	"testing"
	8	"time"
	9
	10	"golang.org/x/crypto/ssh"
	11	)
	12
	13	// execStatus runs cmd on a new session and returns its exit status and
	14	// stderr; -1 when the session could not run.
	15	func execStatus(client *ssh.Client, cmd string) (int, string) {
	16	sess, err := client.NewSession()
	17	if err != nil {
	18	return -1, err.Error()
	19	}
	20	defer sess.Close()
	21	var stderr bytes.Buffer
	22	sess.Stderr = &stderr
	23	err = sess.Run(cmd)
	24	var exit *ssh.ExitError
	25	switch {
	26	case err == nil:
	27	return 0, stderr.String()
	28	case errors.As(err, &exit):
	29	return exit.ExitStatus(), stderr.String()
	30	}
	31	return -1, err.Error()
	32	}
	33
	34	// waitClosed fails unless the server closes the client's connection
	35	// within five seconds.
	36	func waitClosed(t testing.T, client ssh.Client) {
	37	t.Helper()
	38	done := make(chan struct{})
	39	go func() { client.Wait(); close(done) }()
	40	select {
	41	case <-done:
	42	case <-time.After(5 * time.Second):
	43	t.Fatal("the connection stayed open")
	44	}
	45	}
	46
	47	// Each exec reads the key again. The rows change behind the store's
	48	// back here, so no revocation is announced and the connection stays up:
	49	// what refuses the command is the per-exec check alone.
	50	func TestExecRevalidatesKey(t *testing.T) {
	51	ts := newTestServer(t)
	52	if code, errOut := execStatus(ts.client, "whoami"); code != 0 {
	53	t.Fatalf("whoami: %d %s", code, errOut)
	54	}
	55	if _, err := ts.st.DB.Exec("UPDATE ssh_keys SET scope = 'git' WHERE id = ?", ts.keyID); err != nil {
	56	t.Fatal(err)
	57	}
	58	if code, errOut := execStatus(ts.client, "whoami"); code != 4 \|\| !strings.Contains(errOut, "does not allow control commands") {
	59	t.Fatalf("whoami after re-scope: %d %q", code, errOut)
	60	}
	61	if _, err := ts.st.DB.Exec("DELETE FROM ssh_keys WHERE id = ?", ts.keyID); err != nil {
	62	t.Fatal(err)
	63	}
	64	if code, errOut := execStatus(ts.client, "whoami"); code != 4 \|\| !strings.Contains(errOut, "no longer registered") {
	65	t.Fatalf("whoami after delete: %d %q", code, errOut)
	66	}
	67	}
	68
	69	// Removing the key cuts the connection, ending a command running on it.
	70	func TestRemoveKeyCutsConnection(t *testing.T) {
	71	ts := newTestServer(t)
	72	withBuild(t, ts)
	73	var stderr bytes.Buffer
	74	sess := startFollow(t, ts.client, &stderr)
	75	if err := ts.st.RemoveSSHKey(ts.uid, ts.fp); err != nil {
	76	t.Fatal(err)
	77	}
	78	waited := make(chan error, 1)
	79	go func() { waited <- sess.Wait() }()
	80	select {
	81	case err := <-waited:
	82	if err == nil {
	83	t.Fatal("the follow exited cleanly after its key was removed")
	84	}
	85	case <-time.After(5 * time.Second):
	86	t.Fatal("the follow outlived its key")
	87	}
	88	waitClosed(t, ts.client)
	89	}
	90
	91	func TestDisableCutsConnection(t *testing.T) {
	92	ts := newTestServer(t)
	93	if err := ts.st.SetUserDisabled(ts.uid, true); err != nil {
	94	t.Fatal(err)
	95	}
	96	waitClosed(t, ts.client)
	97	}
	98
	99	// A revocation made by another process is not announced here; the
	100	// sweep finds it. A live key survives the sweep.
	101	func TestSweepCutsOutOfProcessRevocation(t *testing.T) {
	102	ts := newTestServer(t)
	103	ts.srv.sweepOnce()
	104	if code, errOut := execStatus(ts.client, "whoami"); code != 0 {
	105	t.Fatalf("the sweep cut a live key: %d %s", code, errOut)
	106	}
	107	if _, err := ts.st.DB.Exec("UPDATE users SET disabled = 1 WHERE id = ?", ts.uid); err != nil {
	108	t.Fatal(err)
	109	}
	110	ts.srv.sweepOnce()
	111	waitClosed(t, ts.client)
	112	}

internal/sshd/sshd.go +121 −17

		@@ -12,9 +12,11 @@ import (
12	12	"fmt"
13	13	"io"
14	14	"log/slog"
	15	"maps"
15	16	"net"
16	17	"os"
17	18	"path/filepath"
	19	"slices"
18	20	"strconv"
19	21	"sync"
20	22	"sync/atomic"
		@@ -50,6 +52,19 @@ type Server struct {
50	52	type conn struct {
51	53	net net.Conn
52	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.
	65	func (c *conn) cut() {
	66	c.cutOnce.Do(func() { close(c.revoked) })
	67	c.net.Close()
53	68	}
54	69
55	70	func New(cfg config.Config, st store.Store) (Server, error) {
		@@ -67,6 +82,7 @@ func New(cfg config.Config, st store.Store) (Server, error) {
67	82	sc.AddHostKey(sg)
68	83	}
69	84	s.sshCfg = sc
	85	st.OnRevoke(s.revoke)
70	86	return s, nil
71	87	}
72	88
		@@ -150,19 +166,20 @@ func (s Server) authenticate(meta ssh.ConnMetadata, pub ssh.PublicKey) (ssh.Pe
150	166	return &ssh.Permissions{Extensions: map[string]string{
151	167	"user-id": strconv.FormatInt(key.UserID, 10),
152	168	"key-id": strconv.FormatInt(key.ID, 10),
153		"key-fp": fp,
154		"scope": key.Scope,
155	169	}}, nil
156	170	}
157	171
158	172	// Serve accepts connections on ln until it is closed.
159	173	func (s *Server) Serve(ln net.Listener) error {
	174	served := make(chan struct{})
	175	defer close(served)
	176	go s.sweep(served)
160	177	for {
161	178	nc, err := ln.Accept()
162	179	if err != nil {
163	180	return err
164	181	}
165		c := &conn{net: nc}
	182	c := &conn{net: nc, revoked: make(chan struct{})}
166	183	s.mu.Lock()
167	184	s.conns[c] = struct{}{}
168	185	s.mu.Unlock()
		@@ -179,6 +196,76 @@ func (s *Server) Serve(ln net.Listener) error {
179	196	}
180	197	}
181	198
	199	// revoke closes the connections opened by the keys r names.
	200	func (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.
	219	const sweepInterval = 15 * time.Second
	220
	221	func (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.
	239	func (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
182	269	// Stop ends the commands that run until something happens (build log
183	270	// --follow), so a shutdown drain waits only for work that finishes. It
184	271	// does not close connections; Shutdown does.
		@@ -218,6 +305,11 @@ func (s Server) handleConn(c conn) {
218	305	return
219	306	}
220	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()
221	313	go ssh.DiscardRequests(reqs)
222	314
223	315	for newCh := range chans {
		@@ -232,12 +324,12 @@ func (s Server) handleConn(c conn) {
232	324	c.active.Add(1)
233	325	go func() {
234	326	defer c.active.Add(-1)
235		s.handleSession(sconn, ch, chReqs)
	327	s.handleSession(c, sconn, ch, chReqs)
236	328	}()
237	329	}
238	330	}
239	331
240		func (s Server) handleSession(sconn ssh.ServerConn, ch ssh.Channel, reqs <-chan *ssh.Request) {
	332	func (s Server) handleSession(c conn, sconn ssh.ServerConn, ch ssh.Channel, reqs <-chan ssh.Request) {
241	333	defer ch.Close()
242	334	var term control.Term
243	335	for req := range reqs {
		@@ -268,7 +360,7 @@ func (s Server) handleSession(sconn ssh.ServerConn, ch ssh.Channel, reqs <-cha
268	360	}
269	361	close(done)
270	362	}()
271		code := s.runExec(sconn, ch, term, payload.Command, done)
	363	code := s.runExec(c, sconn, ch, term, payload.Command, done)
272	364	sendExit(ch, code)
273	365	return
274	366	case "shell":
		@@ -296,20 +388,32 @@ func sendExit(ch ssh.Channel, code int) {
296	388	ch.SendRequest("exit-status", false, ssh.Marshal(&msg))
297	389	}
298	390
299		func (s Server) runExec(sconn ssh.ServerConn, ch ssh.Channel, term control.Term, cmdline string, done <-chan struct{}) int {
	391	func (s Server) runExec(c conn, sconn *ssh.ServerConn, ch ssh.Channel, term control.Term, cmdline string, done <-chan struct{}) int {
300	392	ext := sconn.Permissions.Extensions
301	393	if blob := ext["anon-key"]; blob != "" {
302	394	return s.runAnonymous(ch, blob, cmdline)
303	395	}
304	396	userID, _ := strconv.ParseInt(ext["user-id"], 10, 64)
305	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	}
306	410	user, err := s.st.UserByID(userID)
307	411	if err != nil {
308	412	fmt.Fprintln(ch.Stderr(), "account no longer exists")
309	413	return protocol.ExitDenied
310	414	}
311	415	_ = s.st.TouchSSHKey(keyID)
312		return Exec(s.cfg, s.st, user, ext["scope"], ext["key-fp"], term, cmdline, ch, ch, ch.Stderr(), done, s.stopping)
	416	return Exec(s.cfg, s.st, user, key, term, cmdline, ch, ch, ch.Stderr(), done, s.stopping, c.revoked)
313	417	}
314	418
315	419	// runAnonymous handles a session from an unregistered key: the register
		@@ -338,9 +442,9 @@ func (s *Server) runAnonymous(ch ssh.Channel, keyB64, cmdline string) int {
338	442
339	443	// Exec runs one SSH exec command line for an authenticated key. It is the
340	444	// single dispatch path shared by the embedded listener and the system-sshd
341		// forced command (gitbayd shell).
342		func Exec(cfg config.Config, st *store.Store, user store.User, scope, source string, term control.Term, cmdline string,
343		stdin io.Reader, stdout, stderr io.Writer, done, stopping <-chan struct{}) int {
	445	// forced command (gitbayd shell). Closing revoked kills a git transport.
	446	func 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 {
344	448	if user.Disabled {
345	449	fmt.Fprintln(stderr, "this account is disabled; contact the instance admin")
346	450	return protocol.ExitDenied
		@@ -357,7 +461,7 @@ func Exec(cfg config.Config, st *store.Store, user store.User, scope, source str
357	461	fmt.Fprintln(stderr, "your account is not active yet: verify your email first")
358	462	return protocol.ExitDenied
359	463	}
360		return runGit(cfg, st, user, scope, argv, stdin, stdout, stderr)
	464	return runGit(cfg, st, user, key.Scope, argv, stdin, stdout, stderr, revoked)
361	465	case "git-lfs-authenticate":
362	466	// Part of the git transport, not the control plane: usable by
363	467	// git-scoped and deploy keys, with the transports' access rules.
		@@ -365,13 +469,13 @@ func Exec(cfg config.Config, st *store.Store, user store.User, scope, source str
365	469	fmt.Fprintln(stderr, "your account is not active yet: verify your email first")
366	470	return protocol.ExitDenied
367	471	}
368		return runLFSAuthenticate(cfg, st, user, scope, argv, stdout, stderr)
	472	return runLFSAuthenticate(cfg, st, user, key.Scope, argv, stdout, stderr)
369	473	}
370	474	}
371	475	ctx := &control.Ctx{
372	476	User: user,
373		Scope: scope,
374		Source: source,
	477	Scope: key.Scope,
	478	Source: key.Fingerprint,
375	479	Term: term,
376	480	Store: st,
377	481	Cfg: cfg,
		@@ -386,7 +490,7 @@ func Exec(cfg config.Config, st *store.Store, user store.User, scope, source str
386	490
387	491	// runGit streams a git transport service after access checks.
388	492	func runGit(cfg config.Config, st *store.Store, user store.User, scope string, argv []string,
389		stdin io.Reader, stdout, stderr io.Writer) int {
	493	stdin io.Reader, stdout, stderr io.Writer, revoked <-chan struct{}) int {
390	494	service := argv[0]
391	495	if len(argv) != 2 {
392	496	fmt.Fprintf(stderr, "usage: %s <path>\n", service)
		@@ -461,7 +565,7 @@ func runGit(cfg config.Config, st *store.Store, user store.User, scope string, a
461	565	}
462	566	}
463	567	}
464		if err := gitutil.Transport(service, dir, stdin, stdout, stderr, env, maxPack, nil); err != nil {
	568	if err := gitutil.Transport(service, dir, stdin, stdout, stderr, env, maxPack, revoked); err != nil {
465	569	return protocol.ExitFailure
466	570	}
467	571	return protocol.ExitOK

internal/sshd/sshd_test.go +46 −17

		@@ -18,10 +18,18 @@ import (
18	18	"gitbay.org/gitbay/internal/store"
19	19	)
20	20
21		// followServer starts an embedded server holding alice, her public repo
22		// alice/app and a queued build 1 whose log has one line, and returns it
23		// with a client connected as alice.
24		func followServer(t testing.T) (Server, *ssh.Client) {
	21	// testServer is an embedded server over a fresh store holding alice
	22	// with one full-scope key, and a client connected with that key.
	23	type testServer struct {
	24	srv *Server
	25	st *store.Store
	26	client *ssh.Client
	27	uid int64
	28	keyID int64
	29	fp string
	30	}
	31
	32	func newTestServer(t *testing.T) testServer {
25	33	t.Helper()
26	34	root := t.TempDir()
27	35	st, err := store.Open(filepath.Join(root, "gitbay.db"))
		@@ -36,17 +44,6 @@ func followServer(t testing.T) (Server, *ssh.Client) {
36	44	if err != nil {
37	45	t.Fatal(err)
38	46	}
39		repoID, err := st.CreateRepo("user", uid, "app", "public")
40		if err != nil {
41		t.Fatal(err)
42		}
43		id, err := st.CreateBuild(repoID, "unit", "abc", "main", `["true"]`, "", "", true)
44		if err != nil {
45		t.Fatal(err)
46		}
47		if err := st.AppendBuildLog(id, []byte("queued\n")); err != nil {
48		t.Fatal(err)
49		}
50	47	_, priv, err := ed25519.GenerateKey(rand.Reader)
51	48	if err != nil {
52	49	t.Fatal(err)
		@@ -56,7 +53,12 @@ func followServer(t testing.T) (Server, *ssh.Client) {
56	53	t.Fatal(err)
57	54	}
58	55	pub := signer.PublicKey()
59		if err := st.AddSSHKey(uid, ssh.FingerprintSHA256(pub), pub.Type(), pub.Marshal(), "full", "test"); err != nil {
	56	fp := ssh.FingerprintSHA256(pub)
	57	if err := st.AddSSHKey(uid, fp, pub.Type(), pub.Marshal(), "full", "test"); err != nil {
	58	t.Fatal(err)
	59	}
	60	key, err := st.SSHKeyByFingerprint(fp)
	61	if err != nil {
60	62	t.Fatal(err)
61	63	}
62	64
		@@ -83,7 +85,34 @@ func followServer(t testing.T) (Server, *ssh.Client) {
83	85	t.Fatal(err)
84	86	}
85	87	t.Cleanup(func() { client.Close() })
86		return srv, client
	88	return testServer{srv: srv, st: st, client: client, uid: uid, keyID: key.ID, fp: fp}
	89	}
	90
	91	// withBuild gives alice the public repo alice/app and a queued build 1
	92	// whose log has one line.
	93	func withBuild(t *testing.T, ts testServer) {
	94	t.Helper()
	95	repoID, err := ts.st.CreateRepo("user", ts.uid, "app", "public")
	96	if err != nil {
	97	t.Fatal(err)
	98	}
	99	id, err := ts.st.CreateBuild(repoID, "unit", "abc", "main", `["true"]`, "", "", true)
	100	if err != nil {
	101	t.Fatal(err)
	102	}
	103	if err := ts.st.AppendBuildLog(id, []byte("queued\n")); err != nil {
	104	t.Fatal(err)
	105	}
	106	}
	107
	108	// followServer starts an embedded server holding alice, her public repo
	109	// alice/app and a queued build 1 whose log has one line, and returns it
	110	// with a client connected as alice.
	111	func followServer(t testing.T) (Server, *ssh.Client) {
	112	t.Helper()
	113	ts := newTestServer(t)
	114	withBuild(t, ts)
	115	return ts.srv, ts.client
87	116	}
88	117
89	118	// startFollow runs build log --follow on a new session and returns once

Commit c0969481b0

Commit `c0969481b0`