krz/gitbay

A CLI-first git forge. cli forge git self-hosted https://gitbay.org

Commit 568b879143

568b87914365b4278f02383e9da7756ed7f243ce

parent: 973797757f

Verified · cmc

cmc <hello@cleberg.net> · 2026-09-29 16:31 UTC 

sshd: receive-pack takes a slot from its own push limit

A deploy key is its own principal. The slot covers both hooks and is
released once receive-pack has exited.

Ref #308

Layout: unified · split

cmd/gitbayd/main.go +15 −1
@@ -262,12 +262,20 @@ func serveCmd() *cobra.Command {
262 if packMax > 1 { 262 if packMax > 1 {
263 packs.CapClass("ip:", packMax-1) 263 packs.CapClass("ip:", packMax-1)
264 } 264 }
265 // receive-pack has a budget of its own. Parallel pushes
266 // from one account (scripts, bots, several terminals)
267 // queue, up to half the queue, rather than being refused
268 // once one is waiting.
269 pushMax, pushPer, pushQueue, pushWait := cfg.Limits.PushLimits()
270 pushes := packlimit.New(pushMax, pushPer, pushQueue, pushWait)
271 pushes.Name("push")
272 pushes.CapQueue(pushPerQueue(pushQueue))
265 273
266 errCh := make(chan error, 3) 274 errCh := make(chan error, 3)
267 var sshSrv *sshd.Server 275 var sshSrv *sshd.Server
268 var sshLn, gitLn net.Listener 276 var sshLn, gitLn net.Listener
269 if cfg.SSH.Mode == "embedded" { 277 if cfg.SSH.Mode == "embedded" {
270 srv, err := sshd.New(cfg, st, packs) 278 srv, err := sshd.New(cfg, st, packs, pushes)
271 if err != nil { 279 if err != nil {
272 return err 280 return err
273 } 281 }
@@ -645,3 +653,9 @@ func reapPending(ctx context.Context, st *store.Store, maxAge time.Duration) {
645 } 653 }
646 } 654 }
647} 655}
656
657// pushPerQueue is how many pushes one principal may have waiting: half
658// the queue, at least one.
659func pushPerQueue(queue int) int {
660 return max(1, queue/2)
661}
cmd/gitbayd/system.go +2 −2
@@ -100,8 +100,8 @@ func shellCmd() *cobra.Command {
100 os.Exit(protocol.ExitUsage) 100 os.Exit(protocol.ExitUsage)
101 } 101 }
102 // Each forced command is its own process, so there is no 102 // Each forced command is its own process, so there is no
103 // shared pack budget in system mode. 103 // shared pack or push budget in system mode.
104 code := sshd.Exec(cfg, st, nil, user, key, control.ParseTerm(os.Getenv("GITBAY_TERM")), cmdline, os.Stdin, os.Stdout, os.Stderr, nil, nil, nil) 104 code := sshd.Exec(cfg, st, nil, nil, user, key, control.ParseTerm(os.Getenv("GITBAY_TERM")), cmdline, os.Stdin, os.Stdout, os.Stderr, nil, nil, nil)
105 st.Close() 105 st.Close()
106 os.Exit(code) 106 os.Exit(code)
107 return nil 107 return nil
internal/packlimit/packlimit.go +13 −2
@@ -24,6 +24,7 @@ var (
24 24
25type Limiter struct { 25type Limiter struct {
26 max, per, queue int 26 max, per, queue int
27 perQueue int // waiting, per principal; per unless set
27 wait time.Duration 28 wait time.Duration
28 name string // what is limited, for the refusal log 29 name string // what is limited, for the refusal log
29 30
@@ -47,7 +48,7 @@ func New(max, per, queue int, wait time.Duration) *Limiter {
47 if max <= 0 { 48 if max <= 0 {
48 return nil 49 return nil
49 } 50 }
50 return &Limiter{max: max, per: per, queue: queue, wait: wait, name: "pack", 51 return &Limiter{max: max, per: per, perQueue: per, queue: queue, wait: wait, name: "pack",
51 held: map[string]int{}, waiting: map[string]int{}, changed: make(chan struct{}), 52 held: map[string]int{}, waiting: map[string]int{}, changed: make(chan struct{}),
52 warned: map[string]time.Time{}} 53 warned: map[string]time.Time{}}
53} 54}
@@ -86,6 +87,16 @@ func (l *Limiter) Name(name string) {
86 l.name = name 87 l.name = name
87} 88}
88 89
90// CapQueue lets one principal have up to n requests waiting, where by
91// default it may have as many as it may run. It applies only while a
92// per-principal cap is set. Call it before the limiter is in use.
93func (l *Limiter) CapQueue(n int) {
94 if l == nil {
95 return
96 }
97 l.perQueue = n
98}
99
89// CapClass caps the slots that principals starting with prefix may hold 100// CapClass caps the slots that principals starting with prefix may hold
90// between them. Call it before the limiter is in use. 101// between them. Call it before the limiter is in use.
91func (l *Limiter) CapClass(prefix string, n int) { 102func (l *Limiter) CapClass(prefix string, n int) {
@@ -127,7 +138,7 @@ func (l *Limiter) Acquire(done <-chan struct{}, principal string) (release func(
127 l.mu.Unlock() 138 l.mu.Unlock()
128 return l.releaser(principal), nil 139 return l.releaser(principal), nil
129 } 140 }
130 if l.queued >= l.queue || (l.per > 0 && l.waiting[principal] >= l.per) { 141 if l.queued >= l.queue || (l.per > 0 && l.waiting[principal] >= l.perQueue) {
131 l.mu.Unlock() 142 l.mu.Unlock()
132 return nil, ErrBusy 143 return nil, ErrBusy
133 } 144 }
internal/packlimit/packlimit_test.go +28
@@ -337,3 +337,31 @@ func TestRefusedNamesTheLimit(t *testing.T) {
337 var none *Limiter 337 var none *Limiter
338 none.Name("push") 338 none.Name("push")
339} 339}
340
341// With a waiting cap above per, one principal runs per and queues up to
342// perQueue, and is refused past that.
343func TestPerPrincipalQueueCap(t *testing.T) {
344 l := New(4, 1, 16, 5*time.Second)
345 l.CapQueue(4)
346 r, err := l.Acquire(nil, "a")
347 if err != nil {
348 t.Fatal(err)
349 }
350 for i := 1; i <= 4; i++ {
351 go func() {
352 if r, err := l.Acquire(nil, "a"); err == nil {
353 r()
354 }
355 }()
356 waitQueued(t, l, i)
357 }
358 if _, err := l.Acquire(nil, "a"); !errors.Is(err, ErrBusy) {
359 t.Fatalf("sixth for a: %v", err)
360 }
361 rb, err := l.Acquire(nil, "b")
362 if err != nil {
363 t.Fatalf("b blocked by a's queue: %v", err)
364 }
365 rb()
366 r()
367}
internal/sshd/refusal_test.go +156 −7
@@ -2,9 +2,11 @@ package sshd
2 2
3import ( 3import (
4 "bytes" 4 "bytes"
5 "errors"
5 "io" 6 "io"
6 "os" 7 "os"
7 "path/filepath" 8 "path/filepath"
9 "strconv"
8 "strings" 10 "strings"
9 "testing" 11 "testing"
10 "time" 12 "time"
@@ -56,7 +58,7 @@ func TestRefusedPushIsAudited(t *testing.T) {
56 bob.Pending = pending 58 bob.Pending = pending
57 key := store.SSHKey{Scope: "full", Fingerprint: "SHA256:test"} 59 key := store.SSHKey{Scope: "full", Fingerprint: "SHA256:test"}
58 var out, errOut bytes.Buffer 60 var out, errOut bytes.Buffer
59 code := Exec(cfg, st, nil, bob, key, control.Term{}, "git-receive-pack alice/app", 61 code := Exec(cfg, st, nil, nil, bob, key, control.Term{}, "git-receive-pack alice/app",
60 strings.NewReader(""), &out, &errOut, nil, nil, nil) 62 strings.NewReader(""), &out, &errOut, nil, nil, nil)
61 if code != protocol.ExitDenied { 63 if code != protocol.ExitDenied {
62 t.Fatalf("pending %v: exit %d: %s", pending, code, errOut.String()) 64 t.Fatalf("pending %v: exit %d: %s", pending, code, errOut.String())
@@ -83,7 +85,7 @@ func TestCloneRefusedWhenPackSlotsAreFull(t *testing.T) {
83 key := store.SSHKey{Scope: "full", Fingerprint: "SHA256:test"} 85 key := store.SSHKey{Scope: "full", Fingerprint: "SHA256:test"}
84 for _, service := range []string{"git-upload-pack", "git-upload-archive"} { 86 for _, service := range []string{"git-upload-pack", "git-upload-archive"} {
85 var out, errOut bytes.Buffer 87 var out, errOut bytes.Buffer
86 code := Exec(cfg, st, packs, bob, key, control.Term{}, service+" alice/app", 88 code := Exec(cfg, st, packs, nil, bob, key, control.Term{}, service+" alice/app",
87 strings.NewReader(""), &out, &errOut, nil, nil, nil) 89 strings.NewReader(""), &out, &errOut, nil, nil, nil)
88 if code != protocol.ExitFailure || !strings.Contains(errOut.String(), "busy") { 90 if code != protocol.ExitFailure || !strings.Contains(errOut.String(), "busy") {
89 t.Fatalf("%s: exit %d: %q", service, code, errOut.String()) 91 t.Fatalf("%s: exit %d: %q", service, code, errOut.String())
@@ -106,7 +108,7 @@ func TestPushBypassesPackLimitAndCloneReleasesSlot(t *testing.T) {
106 packs := packlimit.New(1, 0, 0, time.Second) 108 packs := packlimit.New(1, 0, 0, time.Second)
107 109
108 var out, errOut bytes.Buffer 110 var out, errOut bytes.Buffer
109 if code := Exec(cfg, st, packs, alice, key, control.Term{}, "git-upload-pack alice/app", 111 if code := Exec(cfg, st, packs, nil, alice, key, control.Term{}, "git-upload-pack alice/app",
110 strings.NewReader("0000"), &out, &errOut, nil, nil, nil); code != protocol.ExitOK { 112 strings.NewReader("0000"), &out, &errOut, nil, nil, nil); code != protocol.ExitOK {
111 t.Fatalf("clone: exit %d: %s", code, errOut.String()) 113 t.Fatalf("clone: exit %d: %s", code, errOut.String())
112 } 114 }
@@ -118,7 +120,7 @@ func TestPushBypassesPackLimitAndCloneReleasesSlot(t *testing.T) {
118 120
119 out.Reset() 121 out.Reset()
120 errOut.Reset() 122 errOut.Reset()
121 if code := Exec(cfg, st, packs, alice, key, control.Term{}, "git-receive-pack alice/app", 123 if code := Exec(cfg, st, packs, nil, alice, key, control.Term{}, "git-receive-pack alice/app",
122 strings.NewReader("0000"), &out, &errOut, nil, nil, nil); code != protocol.ExitOK { 124 strings.NewReader("0000"), &out, &errOut, nil, nil, nil); code != protocol.ExitOK {
123 t.Fatalf("push with slots full: exit %d: %s", code, errOut.String()) 125 t.Fatalf("push with slots full: exit %d: %s", code, errOut.String())
124 } 126 }
@@ -159,7 +161,7 @@ func killedClone(t *testing.T, stdout io.Writer, done, stopping, revoked <-chan
159 packs := packlimit.New(1, 0, 0, time.Second) 161 packs := packlimit.New(1, 0, 0, time.Second)
160 codec := make(chan int, 1) 162 codec := make(chan int, 1)
161 go func() { 163 go func() {
162 codec <- Exec(cfg, st, packs, alice, key, control.Term{}, "git-upload-pack alice/app", 164 codec <- Exec(cfg, st, packs, nil, alice, key, control.Term{}, "git-upload-pack alice/app",
163 silentStdin(t), stdout, io.Discard, done, stopping, revoked) 165 silentStdin(t), stdout, io.Discard, done, stopping, revoked)
164 }() 166 }()
165 select { 167 select {
@@ -208,7 +210,7 @@ func TestCloneRunsOnDuringRestart(t *testing.T) {
208 key := store.SSHKey{Scope: "full", Fingerprint: "SHA256:test"} 210 key := store.SSHKey{Scope: "full", Fingerprint: "SHA256:test"}
209 packs := packlimit.New(1, 0, 0, time.Second) 211 packs := packlimit.New(1, 0, 0, time.Second)
210 var errOut bytes.Buffer 212 var errOut bytes.Buffer
211 if code := Exec(cfg, st, packs, alice, key, control.Term{}, "git-upload-pack alice/app", 213 if code := Exec(cfg, st, packs, nil, alice, key, control.Term{}, "git-upload-pack alice/app",
212 strings.NewReader("0000"), io.Discard, &errOut, closed(), closed(), nil); code != protocol.ExitOK { 214 strings.NewReader("0000"), io.Discard, &errOut, closed(), closed(), nil); code != protocol.ExitOK {
213 t.Fatalf("exit %d: %s", code, errOut.String()) 215 t.Fatalf("exit %d: %s", code, errOut.String())
214 } 216 }
@@ -233,9 +235,156 @@ func TestRefusedCloneStaysOffLimiter(t *testing.T) {
233 defer hold() 235 defer hold()
234 key := store.SSHKey{Scope: "full", Fingerprint: "SHA256:test"} 236 key := store.SSHKey{Scope: "full", Fingerprint: "SHA256:test"}
235 var out, errOut bytes.Buffer 237 var out, errOut bytes.Buffer
236 code := Exec(cfg, st, packs, bob, key, control.Term{}, "git-upload-pack alice/secret", 238 code := Exec(cfg, st, packs, nil, bob, key, control.Term{}, "git-upload-pack alice/secret",
237 strings.NewReader(""), &out, &errOut, nil, nil, nil) 239 strings.NewReader(""), &out, &errOut, nil, nil, nil)
238 if code != protocol.ExitNotFound || strings.Contains(errOut.String(), "busy") { 240 if code != protocol.ExitNotFound || strings.Contains(errOut.String(), "busy") {
239 t.Fatalf("exit %d: %q", code, errOut.String()) 241 t.Fatalf("exit %d: %q", code, errOut.String())
240 } 242 }
241} 243}
244
245// hungUpStdin is a client that sends nothing until hangUp, which ends
246// its stdin the way a closed channel does.
247func hungUpStdin(t *testing.T) (r *os.File, hangUp func()) {
248 t.Helper()
249 r, w, err := os.Pipe()
250 if err != nil {
251 t.Fatal(err)
252 }
253 t.Cleanup(func() { r.Close(); w.Close() })
254 return r, func() { w.Close() }
255}
256
257// queuedFor waits until principal has a push waiting on l: a probe that
258// gives up at once is then refused busy rather than queued.
259func queuedFor(t *testing.T, l *packlimit.Limiter, principal string) {
260 t.Helper()
261 deadline := time.Now().Add(5 * time.Second)
262 for time.Now().Before(deadline) {
263 if _, err := l.Acquire(closed(), principal); errors.Is(err, packlimit.ErrBusy) {
264 return
265 }
266 time.Sleep(10 * time.Millisecond)
267 }
268 t.Fatalf("no push queued for %s", principal)
269}
270
271// With push_per_principal 1 one account runs one push, queues a second
272// and is refused a third, while another principal still gets in. A
273// client hanging up frees its slot for the one queued behind it.
274func TestPushPerPrincipalCap(t *testing.T) {
275 cfg, st, alice := cloneFixture(t)
276 repo, err := st.RepoByPath("alice/app")
277 if err != nil {
278 t.Fatal(err)
279 }
280 key := store.SSHKey{ID: 1, Scope: "full", Fingerprint: "SHA256:test"}
281 pushes := packlimit.New(2, 1, 16, 10*time.Second)
282 push := func(k store.SSHKey, stdin io.Reader, errOut io.Writer) <-chan int {
283 codec := make(chan int, 1)
284 go func() {
285 codec <- Exec(cfg, st, nil, pushes, alice, k, control.Term{}, "git-receive-pack alice/app",
286 stdin, io.Discard, errOut, nil, nil, nil)
287 }()
288 return codec
289 }
290 exited := func(codec <-chan int, want int, what string) {
291 t.Helper()
292 select {
293 case code := <-codec:
294 if code != want {
295 t.Fatalf("%s: exit %d", what, code)
296 }
297 case <-time.After(5 * time.Second):
298 t.Fatalf("%s still running", what)
299 }
300 }
301 principal := "user:" + strconv.FormatInt(alice.ID, 10)
302
303 in1, hangUp1 := hungUpStdin(t)
304 first := push(key, in1, io.Discard)
305 // The first holds alice's one slot once a probe cannot take it.
306 deadline := time.Now().Add(5 * time.Second)
307 for {
308 r, err := pushes.Acquire(closed(), principal)
309 if err != nil {
310 break
311 }
312 r()
313 if time.Now().After(deadline) {
314 t.Fatal("first push never took a slot")
315 }
316 time.Sleep(10 * time.Millisecond)
317 }
318 in2, hangUp2 := hungUpStdin(t)
319 second := push(key, in2, io.Discard)
320 queuedFor(t, pushes, principal)
321
322 var errOut bytes.Buffer
323 if code := Exec(cfg, st, nil, pushes, alice, key, control.Term{}, "git-receive-pack alice/app",
324 strings.NewReader(""), io.Discard, &errOut, nil, nil, nil); code != protocol.ExitFailure ||
325 !strings.Contains(errOut.String(), "limit of concurrent pushes") {
326 t.Fatalf("third push: exit %d: %q", code, errOut.String())
327 }
328
329 // A deploy key on the same account is its own principal: it takes
330 // the second global slot while alice's push waits.
331 deploy := store.SSHKey{ID: 2, Scope: "deploy:" + strconv.FormatInt(repo.ID, 10) + ":rw", Fingerprint: "SHA256:deploy"}
332 exited(push(deploy, strings.NewReader("0000"), io.Discard), protocol.ExitOK, "deploy key push")
333
334 // receive-pack fails a client that hangs up before sending anything.
335 hangUp1()
336 exited(first, protocol.ExitFailure, "first push")
337 hangUp2()
338 exited(second, protocol.ExitFailure, "second push")
339 for _, p := range []string{"a", "b"} {
340 r, err := pushes.Acquire(nil, p)
341 if err != nil {
342 t.Fatalf("slot not released: %v", err)
343 }
344 defer r()
345 }
346}
347
348// A revoked key kills a push waiting on its client, and the slot comes
349// back.
350func TestPushKilledWhenKeyRevokedReleasesSlot(t *testing.T) {
351 cfg, st, alice := cloneFixture(t)
352 key := store.SSHKey{ID: 1, Scope: "full", Fingerprint: "SHA256:test"}
353 pushes := packlimit.New(1, 1, 0, time.Second)
354 codec := make(chan int, 1)
355 go func() {
356 codec <- Exec(cfg, st, nil, pushes, alice, key, control.Term{}, "git-receive-pack alice/app",
357 silentStdin(t), io.Discard, io.Discard, nil, nil, closed())
358 }()
359 select {
360 case code := <-codec:
361 if code != protocol.ExitFailure {
362 t.Fatalf("exit %d, want the push killed", code)
363 }
364 case <-time.After(5 * time.Second):
365 t.Fatal("push still running")
366 }
367 r, err := pushes.Acquire(nil, "elsewhere")
368 if err != nil {
369 t.Fatalf("slot not released after the kill: %v", err)
370 }
371 r()
372}
373
374// A push the key may not make is refused before it reaches the limiter.
375func TestRefusedPushStaysOffLimiter(t *testing.T) {
376 cfg, st, bob := execFixture(t)
377 pushes := packlimit.New(1, 0, 0, time.Second)
378 hold, err := pushes.Acquire(nil, "elsewhere")
379 if err != nil {
380 t.Fatal(err)
381 }
382 defer hold()
383 key := store.SSHKey{Scope: "full", Fingerprint: "SHA256:test"}
384 var errOut bytes.Buffer
385 code := Exec(cfg, st, nil, pushes, bob, key, control.Term{}, "git-receive-pack alice/app",
386 strings.NewReader(""), io.Discard, &errOut, nil, nil, nil)
387 if code != protocol.ExitDenied || strings.Contains(errOut.String(), "busy") {
388 t.Fatalf("exit %d: %q", code, errOut.String())
389 }
390}
internal/sshd/sshd.go +50 −22
@@ -39,6 +39,7 @@ type Server struct {
39 cfg config.Config 39 cfg config.Config
40 st *store.Store 40 st *store.Store
41 packs *packlimit.Limiter 41 packs *packlimit.Limiter
42 pushes *packlimit.Limiter
42 sshCfg *ssh.ServerConfig 43 sshCfg *ssh.ServerConfig
43 authLimiter *rateLimiter 44 authLimiter *rateLimiter
44 sessions sync.WaitGroup // accepted connections still being served 45 sessions sync.WaitGroup // accepted connections still being served
@@ -70,8 +71,8 @@ func (c *conn) cut() {
70 c.net.Close() 71 c.net.Close()
71} 72}
72 73
73func New(cfg config.Config, st *store.Store, packs *packlimit.Limiter) (*Server, error) { 74func New(cfg config.Config, st *store.Store, packs, pushes *packlimit.Limiter) (*Server, error) {
74 s := &Server{cfg: cfg, st: st, packs: packs, authLimiter: newRateLimiter(cfg.Limits.SSHAuthRate, time.Minute), conns: map[*conn]struct{}{}, stopping: make(chan struct{})} 75 s := &Server{cfg: cfg, st: st, packs: packs, pushes: pushes, authLimiter: newRateLimiter(cfg.Limits.SSHAuthRate, time.Minute), conns: map[*conn]struct{}{}, stopping: make(chan struct{})}
75 76
76 sc := &ssh.ServerConfig{ 77 sc := &ssh.ServerConfig{
77 PublicKeyCallback: s.authenticate, 78 PublicKeyCallback: s.authenticate,
@@ -425,7 +426,7 @@ func (s *Server) runExec(c *conn, sconn *ssh.ServerConn, ch ssh.Channel, term co
425 return protocol.ExitDenied 426 return protocol.ExitDenied
426 } 427 }
427 _ = s.st.TouchSSHKey(keyID) 428 _ = s.st.TouchSSHKey(keyID)
428 return Exec(s.cfg, s.st, s.packs, user, key, term, cmdline, ch, ch, ch.Stderr(), done, s.stopping, c.revoked) 429 return Exec(s.cfg, s.st, s.packs, s.pushes, user, key, term, cmdline, ch, ch, ch.Stderr(), done, s.stopping, c.revoked)
429} 430}
430 431
431// runAnonymous handles a session from an unregistered key: the register 432// runAnonymous handles a session from an unregistered key: the register
@@ -461,7 +462,9 @@ func (s *Server) runAnonymous(ch ssh.Channel, keyB64, cmdline string) int {
461// Exec runs one SSH exec command line for an authenticated key. It is the 462// Exec runs one SSH exec command line for an authenticated key. It is the
462// single dispatch path shared by the embedded listener and the system-sshd 463// single dispatch path shared by the embedded listener and the system-sshd
463// forced command (gitbayd shell). Closing revoked kills a git transport. 464// forced command (gitbayd shell). Closing revoked kills a git transport.
464func Exec(cfg config.Config, st *store.Store, packs *packlimit.Limiter, user store.User, key store.SSHKey, term control.Term, cmdline string, 465// packs bounds clones and fetches, and repo download; pushes bounds
466// receive-pack. A nil limiter is no limit.
467func Exec(cfg config.Config, st *store.Store, packs, pushes *packlimit.Limiter, user store.User, key store.SSHKey, term control.Term, cmdline string,
465 stdin io.Reader, stdout, stderr io.Writer, done, stopping, revoked <-chan struct{}) int { 468 stdin io.Reader, stdout, stderr io.Writer, done, stopping, revoked <-chan struct{}) int {
466 if user.Disabled { 469 if user.Disabled {
467 fmt.Fprintln(stderr, "this account is disabled; contact the instance admin") 470 fmt.Fprintln(stderr, "this account is disabled; contact the instance admin")
@@ -479,7 +482,7 @@ func Exec(cfg config.Config, st *store.Store, packs *packlimit.Limiter, user sto
479 if user.Pending { 482 if user.Pending {
480 fmt.Fprintln(stderr, "your account is not active yet: verify your email first") 483 fmt.Fprintln(stderr, "your account is not active yet: verify your email first")
481 } else { 484 } else {
482 code = runGit(cfg, st, packs, user, key.Scope, argv, stdin, stdout, stderr, done, stopping, revoked) 485 code = runGit(cfg, st, packs, pushes, user, key, argv, stdin, stdout, stderr, done, stopping, revoked)
483 } 486 }
484 // A refused push is a refused write, audited like one. runGit 487 // A refused push is a refused write, audited like one. runGit
485 // refuses only with the path as the one argument, so argv[1:] 488 // refuses only with the path as the one argument, so argv[1:]
@@ -517,9 +520,9 @@ func Exec(cfg config.Config, st *store.Store, packs *packlimit.Limiter, user sto
517} 520}
518 521
519// runGit streams a git transport service after access checks. 522// runGit streams a git transport service after access checks.
520func runGit(cfg config.Config, st *store.Store, packs *packlimit.Limiter, user store.User, scope string, argv []string, 523func runGit(cfg config.Config, st *store.Store, packs, pushes *packlimit.Limiter, user store.User, key store.SSHKey, argv []string,
521 stdin io.Reader, stdout, stderr io.Writer, done, stopping, revoked <-chan struct{}) int { 524 stdin io.Reader, stdout, stderr io.Writer, done, stopping, revoked <-chan struct{}) int {
522 service := argv[0] 525 service, scope := argv[0], key.Scope
523 if len(argv) != 2 { 526 if len(argv) != 2 {
524 fmt.Fprintf(stderr, "usage: %s <path>\n", service) 527 fmt.Fprintf(stderr, "usage: %s <path>\n", service)
525 return protocol.ExitUsage 528 return protocol.ExitUsage
@@ -594,6 +597,24 @@ func runGit(cfg config.Config, st *store.Store, packs *packlimit.Limiter, user s
594 } 597 }
595 } 598 }
596 if write { 599 if write {
600 // Pushes have their own budget, so a clone storm cannot starve
601 // them or the reverse. The slot covers receive-pack and both
602 // hooks: git waits for post-receive (RefsUpdated) before it
603 // exits, and that work is the push's cost. A deploy key is its
604 // own principal, not the account that registered it.
605 principal := "user:" + strconv.FormatInt(user.ID, 10)
606 if policy.IsDeployScope(scope) {
607 principal = "key:" + strconv.FormatInt(key.ID, 10)
608 }
609 release, code := takeSlot(pushes, principal, done, stderr,
610 "the server is busy: it is at its limit of concurrent pushes; try again in a minute")
611 if code != protocol.ExitOK {
612 return code
613 }
614 // Deferred before Transport runs, so it fires after receive-pack
615 // has exited, on every path: the client hanging up ends its
616 // stdin and receive-pack with it, and a revoked key kills it.
617 defer release()
597 // hookd answers only a hook that names this receive-pack. 618 // hookd answers only a hook that names this receive-pack.
598 token, err := st.CreatePushToken(repo.ID, user.ID, scope) 619 token, err := st.CreatePushToken(repo.ID, user.ID, scope)
599 if err != nil { 620 if err != nil {
@@ -606,21 +627,10 @@ func runGit(cfg config.Config, st *store.Store, packs *packlimit.Limiter, user s
606 cancel := revoked 627 cancel := revoked
607 if !write { 628 if !write {
608 // Pack generation shares one budget with smart HTTP and git://. 629 // Pack generation shares one budget with smart HTTP and git://.
609 // receive-pack stays outside it: its post-receive runs after the 630 release, code := takeSlot(packs, "user:"+strconv.FormatInt(user.ID, 10), done, stderr,
610 // client has its report, and must not be queued or killed. 631 "the server is busy: it is at its limit of concurrent clones and fetches; try again in a minute")
611 principal := "user:" + strconv.FormatInt(user.ID, 10) 632 if code != protocol.ExitOK {
612 release, err := packs.Acquire(done, principal) 633 return code
613 if err != nil {
614 packs.Refused("ssh", principal, err)
615 }
616 if errors.Is(err, packlimit.ErrBusy) {
617 fmt.Fprintln(stderr, "the server is busy: it is at its limit of concurrent clones and fetches; try again in a minute")
618 return protocol.ExitFailure
619 }
620 if err != nil {
621 // ErrGone: the client left, or the server is restarting.
622 fmt.Fprintln(stderr, "the server is restarting; try again in a minute")
623 return protocol.ExitFailure
624 } 634 }
625 // Deferred before Transport runs, so it fires after git has 635 // Deferred before Transport runs, so it fires after git has
626 // exited and been waited for. 636 // exited and been waited for.
@@ -673,3 +683,21 @@ func runGit(cfg config.Config, st *store.Store, packs *packlimit.Limiter, user s
673 } 683 }
674 return protocol.ExitOK 684 return protocol.ExitOK
675} 685}
686
687// takeSlot takes a slot from l for principal, waiting until done closes
688// at most. On a refusal it prints busy, or that the server is going
689// away, and returns a nonzero exit.
690func takeSlot(l *packlimit.Limiter, principal string, done <-chan struct{}, stderr io.Writer, busy string) (release func(), code int) {
691 release, err := l.Acquire(done, principal)
692 if err == nil {
693 return release, protocol.ExitOK
694 }
695 l.Refused("ssh", principal, err)
696 if errors.Is(err, packlimit.ErrBusy) {
697 fmt.Fprintln(stderr, busy)
698 } else {
699 // ErrGone: the client left, or the server is restarting.
700 fmt.Fprintln(stderr, "the server is restarting; try again in a minute")
701 }
702 return nil, protocol.ExitFailure
703}
internal/sshd/sshd_test.go +2 −2
@@ -64,7 +64,7 @@ func newTestServer(t *testing.T) testServer {
64 64
65 cfg := config.Default() 65 cfg := config.Default()
66 cfg.Server.Root = root 66 cfg.Server.Root = root
67 srv, err := New(cfg, st, nil) 67 srv, err := New(cfg, st, nil, nil)
68 if err != nil { 68 if err != nil {
69 t.Fatal(err) 69 t.Fatal(err)
70 } 70 }
@@ -245,7 +245,7 @@ func TestUnregisteredKeyMessageNamesFingerprintAndHost(t *testing.T) {
245 // The settings link keeps the site URL's scheme and port. 245 // The settings link keeps the site URL's scheme and port.
246 cfg.Server.SiteURL = "http://forge.test:8080/" 246 cfg.Server.SiteURL = "http://forge.test:8080/"
247 cfg.Registration.Mode = "open" 247 cfg.Registration.Mode = "open"
248 srv, err := New(cfg, st, nil) 248 srv, err := New(cfg, st, nil, nil)
249 if err != nil { 249 if err != nil {
250 t.Fatal(err) 250 t.Fatal(err)
251 } 251 }