From 5219939170c73037600d6dc73411874108d9c13d Mon Sep 17 00:00:00 2001 From: Kristoffer Dalby Date: Sat, 17 Oct 2026 09:35:53 +0000 Subject: [PATCH] types, auth, noise: let one request wait on each auth session One verdict per session, so one waiter: the newest. A displaced one returns at once (register error, SSH re-delegate), not at session end. --- hscontrol/auth.go | 8 ++++ hscontrol/auth_test.go | 83 +++++++++++++++++++++++++++++++++++++++ hscontrol/noise.go | 11 +++++- hscontrol/noise_test.go | 68 ++++++++++++++++++++++++++++++++ hscontrol/types/common.go | 20 ++++++++++ 5 files changed, 188 insertions(+), 2 deletions(-) diff --git a/hscontrol/auth.go b/hscontrol/auth.go index fcba8ecbc..3fc6b089d 100644 --- a/hscontrol/auth.go +++ b/hscontrol/auth.go @@ -342,6 +342,8 @@ func (h *Headscale) waitForFollowup( ) } + superseded := reg.Park() + var verdict types.AuthVerdict select { // Prefer a completed registration even if the context has also @@ -353,6 +355,12 @@ func (h *Headscale) waitForFollowup( select { case <-ctx.Done(): return nil, NewHTTPError(http.StatusUnauthorized, "registration timed out", ctx.Err()) + case <-superseded: + // tailscaled keeps one follow-up in flight, so only a request + // it already abandoned is displaced. + return nil, NewHTTPError( + http.StatusConflict, "superseded by a newer followup request", nil, + ) case verdict = <-reg.WaitForAuth(): } } diff --git a/hscontrol/auth_test.go b/hscontrol/auth_test.go index 040062275..4d2d5dc8e 100644 --- a/hscontrol/auth_test.go +++ b/hscontrol/auth_test.go @@ -4690,3 +4690,86 @@ func TestPendingRegistrationKeepsNoRoutes(t *testing.T) { assert.Equal(t, want, entry.RegistrationData().Hostinfo) assert.Len(t, hostinfo.RoutableIPs, 1, "the request itself must be left alone") } + +// TestFollowupParksOneRequestPerRegistration pins that a pending registration +// keeps one waiting follow-up, the newest. The session hands out one verdict, +// so older waiters are answered at once instead of staying parked until the +// session ends and then each opening a new one. +func TestFollowupParksOneRequestPerRegistration(t *testing.T) { + app := createTestApp(t) + + machineKey := key.NewMachine().Public() + authID := types.MustAuthID() + reg := types.NewRegisterAuthRequest(&types.RegistrationData{ + MachineKey: machineKey, + Hostname: "followup-park-node", + }) + app.state.SetAuthCacheEntry(authID, reg) + + req := tailcfg.RegisterRequest{ + NodeKey: key.NewNode().Public(), + Followup: fmt.Sprintf("http://localhost:8080/register/%s", authID), + } + + const followups = 50 + + type answer struct { + resp *tailcfg.RegisterResponse + err error + } + + answered := make(chan answer, followups) + + for range followups { + go func() { + resp, err := app.waitForFollowup(t.Context(), req, machineKey) + answered <- answer{resp, err} + }() + } + + next := func(msg string) answer { + t.Helper() + + select { + case a := <-answered: + return a + case <-time.After(5 * time.Second): + require.FailNow(t, msg) + + return answer{} + } + } + + for range followups - 1 { + var httpErr HTTPError + require.ErrorAs(t, + next("more than one follow-up is parked on the registration").err, + &httpErr) + assert.Equal(t, http.StatusConflict, httpErr.Code) + } + + select { + case a := <-answered: + require.FailNowf(t, "the newest follow-up must still be waiting", "got %+v", a) + case <-time.After(50 * time.Millisecond): + } + + cacheLen := func() int { + n, _ := app.state.DebugRegistrationCache()["current_len"].(int) + + return n + } + before := cacheLen() + + user := app.state.CreateUserForTest("followup-park-user") + node := app.state.CreateNodeForTest(user, "followup-park-node") + node.MachineKey = machineKey + node.User = user + reg.FinishAuth(types.AuthVerdict{Node: node.View()}) + + last := next("the parked follow-up did not collect the verdict") + require.NoError(t, last.err) + assert.True(t, last.resp.MachineAuthorized) + assert.LessOrEqual(t, cacheLen(), before+1, + "the verdict must not open a session per waiter") +} diff --git a/hscontrol/noise.go b/hscontrol/noise.go index a32ce750a..d6b8a35aa 100644 --- a/hscontrol/noise.go +++ b/hscontrol/noise.go @@ -598,8 +598,9 @@ func (ns *noiseServer) sshActionHoldAndDelegate( } // sshActionFollowUp handles follow-up requests where the client -// provides an auth_id. It blocks until the auth session resolves or -// the request context is cancelled (e.g. the client disconnects). +// provides an auth_id. It blocks until the auth session resolves, the +// request context is cancelled (e.g. the client disconnects), or a newer +// follow-up parks on the session. func (ns *noiseServer) sshActionFollowUp( ctx context.Context, reqLog zerolog.Logger, @@ -663,6 +664,8 @@ func (ns *noiseServer) sshActionFollowUp( reqLog.Trace().Caller().Msg("SSH action follow-up") + superseded := auth.Park() + var ( verdict types.AuthVerdict verdictOK bool @@ -680,6 +683,10 @@ func (ns *noiseServer) sshActionFollowUp( "ssh action follow-up cancelled", ctx.Err(), ) + case <-superseded: + // The newer request collects the verdict; this one re-delegates, as + // it would on a consumed verdict, without waiting for it. + return sessionGone("SSH check follow-up superseded; re-delegating") case verdict, verdictOK = <-auth.WaitForAuth(): } diff --git a/hscontrol/noise_test.go b/hscontrol/noise_test.go index 71c1c8044..83914174e 100644 --- a/hscontrol/noise_test.go +++ b/hscontrol/noise_test.go @@ -2041,3 +2041,71 @@ func TestNoiseRouter_DoesNotServeMetrics(t *testing.T) { assert.Equal(t, http.StatusNotFound, resp.StatusCode) } + +// TestSSHActionFollowUp_ParksOneRequestPerSession pins that a pending SSH +// check keeps one waiting follow-up, the newest. Older waiters re-delegate at +// once instead of staying parked until the session ends. +func TestSSHActionFollowUp_ParksOneRequestPerSession(t *testing.T) { + t.Parallel() + + app := createTestApp(t) + nodes := putSSHCheckNodes(t, app, "ssh-park-user", "src-node", "dst-node") + src, dst := nodes[0], nodes[1] + + authID := types.MustAuthID() + auth := types.NewSSHCheckAuthRequest(src.ID, dst.ID) + app.state.SetAuthCacheEntry(authID, auth) + + ns := &noiseServer{headscale: app, machineKey: dst.MachineKey} + + const followups = 50 + + type answer struct { + action *tailcfg.SSHAction + err error + } + + answered := make(chan answer, followups) + + for range followups { + go func() { + action, err := ns.sshActionFollowUp( + t.Context(), zerolog.Nop(), &tailcfg.SSHAction{}, + authID.String(), src.ID, dst.ID, sshTestLocalUser, + ) + answered <- answer{action, err} + }() + } + + next := func(msg string) answer { + t.Helper() + + select { + case a := <-answered: + return a + case <-time.After(5 * time.Second): + require.FailNow(t, msg) + + return answer{} + } + } + + for range followups - 1 { + a := next("more than one follow-up is parked on the session") + require.NoError(t, a.err) + assert.NotEqual(t, authID, requireSSHHold(t, *a.action), + "re-delegation must mint a new session") + } + + select { + case a := <-answered: + require.FailNowf(t, "the newest follow-up must still be waiting", "got %+v", a) + case <-time.After(50 * time.Millisecond): + } + + auth.FinishAuth(types.AuthVerdict{}) + + last := next("the parked follow-up did not collect the verdict") + require.NoError(t, last.err) + assert.True(t, last.action.Accept, "accept, got %+v", last.action) +} diff --git a/hscontrol/types/common.go b/hscontrol/types/common.go index 69cd1279e..05edd706f 100644 --- a/hscontrol/types/common.go +++ b/hscontrol/types/common.go @@ -143,6 +143,10 @@ type AuthRequest struct { finished chan AuthVerdict closed *atomic.Bool + + // waiter is closed once a newer request parks on this session; see + // [AuthRequest.Park]. + waiter atomic.Pointer[chan struct{}] } // NewAuthRequest creates a pending auth request with no payload, suitable @@ -245,6 +249,22 @@ func (rn *AuthRequest) WaitForAuth() <-chan AuthVerdict { return rn.finished } +// Park makes the caller the one request waiting on [AuthRequest.WaitForAuth] +// and returns a channel that is closed once a newer caller takes its place. +// [AuthRequest.FinishAuth] hands out one verdict, so a session needs one +// waiter. The newest wins, so a client that reconnected is not kept out by +// its own dead request. Parked requests are then bounded by the sessions in +// the auth cache, not by how many requests clients send. +func (rn *AuthRequest) Park() <-chan struct{} { + waiter := make(chan struct{}) + + if prev := rn.waiter.Swap(&waiter); prev != nil { + close(*prev) + } + + return waiter +} + type AuthVerdict struct { // Err is the error that occurred during the authentication process, if any. // If Err is nil, the authentication process has succeeded.