fix(store): passkey 挑战与账号迁移的并发开始改由 advisory 锁串行，登录挑战按来源上限在并发下精确，迁移重启不再给已退役账号留活动记录，补 pgint 契约测试 (0a66385d) · Commits · FelisMC / Felis · GitLab

internal/api/pgrepo.go

+61 −25

Changes for internal/api/pgrepo.go: 61 added lines, 25 removed lines.
Original line number	Diff line number	Diff line
		@@ -1247,6 +1247,9 @@ func (p *PGRepo) SetSetting(ctx context.Context, key string, value []byte) error
		// behind. That bounds the table at one row per (user, purpose): the begin→finish loop
		// nets zero growth, since each begin sweeps the consumed row the previous finish stamped.
		// (Deleting a consumed row is safe: it has already been redeemed and nothing reads it.)
		// Begins for one (user, purpose) take a transaction-scoped advisory lock first: without
		// it two racing begins each delete what the other has not committed yet and both insert,
		// leaving extra rows that only the next begin sweeps.
		// The opaque SessionData is held server-side so the client cannot forge the challenge
		// it must answer at finish.
		func (p *PGRepo) CreatePasskeyChallenge(ctx context.Context, id, userID, purpose string, sessionData []byte, expiresAt time.Time) error {
		@@ -1256,6 +1259,11 @@ func (p *PGRepo) CreatePasskeyChallenge(ctx context.Context, id, userID, purpose
		}
		defer tx.Rollback() //nolint:errcheck // no-op after commit

		if _, err := tx.ExecContext(ctx,
		`SELECT pg_advisory_xact_lock(hashtext('passkey-challenge:' \|\| $1 \|\| ':' \|\| $2))`,
		userID, purpose); err != nil {
		return fmt.Errorf("lock passkey challenges: %w", err)
		}
		if _, err := tx.ExecContext(ctx,
		`DELETE FROM webauthn_challenges WHERE user_id = $1 AND purpose = $2`,
		userID, purpose); err != nil {
		@@ -1316,16 +1324,27 @@ func (p *PGRepo) ConsumePasskeyChallengeByUser(ctx context.Context, userID, purp
		// address or IPv6 /48, see challengeSource) holds in each login store. A ceremony
		// takes seconds and a challenge lives passkeyChallengeTTL, so a network with a few
		// dozen people signing in at once stays well inside it, while a flood of begins
		// fills its own allowance and leaves the other networks their sign-ins. The count
		// and the insert are not serialised, so a burst of truly concurrent begins can pass
		// it together; the per-source token bucket in front of the door caps that burst.
		// fills its own allowance and leaves the other networks their sign-ins. Each store's
		// begins from one source serialise on a transaction-scoped advisory lock
		// (lockChallengeSource), so racing begins cannot all pass one count.
		const maxLiveChallengesPerSource = 32

		// lockChallengeSource takes the advisory lock that makes a login store's per-source
		// count and insert one decision. store names the table, so the two stores' allowances
		// for one source never wait on each other. The lock is released at commit or rollback.
		func lockChallengeSource(ctx context.Context, tx *sql.Tx, store, source string) error {
		if _, err := tx.ExecContext(ctx,
		`SELECT pg_advisory_xact_lock(hashtext($1 \|\| ':' \|\| $2))`, store, source); err != nil {
		return fmt.Errorf("lock %s source: %w", store, err)
		}
		return nil
		}

		// AddPasskeyLoginChallenge stores an email-first login ceremony beside the ones
		// already live for (user, purpose); finish finds it by the challenge the browser
		// signed (ConsumePasskeyLoginChallenge), so a begin by anyone who knows the address
		// never cancels its owner's ceremony. It reaps the account's spent login rows,
		// refuses with ErrTooManyPasskeyChallenges once source holds
		// never cancels its owner's ceremony. Under the source's lock it reaps the account's
		// spent login rows, refuses with ErrTooManyPasskeyChallenges once source holds
		// maxLiveChallengesPerSource live login challenges, then inserts.
		func (p *PGRepo) AddPasskeyLoginChallenge(ctx context.Context, id, userID, purpose, source, challenge string, sessionData []byte, now, expiresAt time.Time) error {
		tx, err := p.db.BeginTx(ctx, nil)
		@@ -1334,6 +1353,9 @@ func (p *PGRepo) AddPasskeyLoginChallenge(ctx context.Context, id, userID, purpo
		}
		defer tx.Rollback() //nolint:errcheck // no-op after commit

		if err := lockChallengeSource(ctx, tx, "webauthn_challenges", source); err != nil {
		return err
		}
		if _, err := tx.ExecContext(ctx,
		`DELETE FROM webauthn_challenges
		WHERE user_id = $1 AND purpose = $2 AND (expires_at <= $3 OR consumed_at IS NOT NULL)`,
		@@ -1408,12 +1430,14 @@ func (p *PGRepo) ConsumePasskeyLoginChallenge(ctx context.Context, userID, purpo
		const maxLiveDiscoverableChallenges = 16384

		// CreateDiscoverableChallenge stashes a discoverable-login ceremony under an opaque handle,
		// bounding the table in one transaction (see the Repo interface for the full contract). It
		// reaps expired/consumed rows first, then refuses when source already holds
		// maxLiveChallengesPerSource live rows or the table holds maxLiveDiscoverableChallenges.
		// Because the reap ran first, the counts are exactly the live rows, so the bounds hold under
		// an adversarial begin-flood (which a reap alone cannot: a burst inside the TTL leaves every
		// fresh row live).
		// bounding the table in one transaction (see the Repo interface for the full contract).
		// Under the source's lock it reaps expired/consumed rows, then refuses when source already
		// holds maxLiveChallengesPerSource live rows or the table holds
		// maxLiveDiscoverableChallenges. Because the reap ran first, the counts are exactly the live
		// rows, so the bounds hold under an adversarial begin-flood (which a reap alone cannot: a
		// burst inside the TTL leaves every fresh row live). The per-source bound is exact; the
		// table bound can be passed by begins from different sources racing the same count, at
		// most one row per open connection, which leaves the few-MB ceiling where it was.
		func (p *PGRepo) CreateDiscoverableChallenge(ctx context.Context, id, source string, sessionData []byte, now, expiresAt time.Time) error {
		tx, err := p.db.BeginTx(ctx, nil)
		if err != nil {
		@@ -1421,6 +1445,9 @@ func (p *PGRepo) CreateDiscoverableChallenge(ctx context.Context, id, source str
		}
		defer tx.Rollback() //nolint:errcheck // no-op after commit

		if err := lockChallengeSource(ctx, tx, "webauthn_discoverable_challenges", source); err != nil {
		return err
		}
		if _, err := tx.ExecContext(ctx,
		`DELETE FROM webauthn_discoverable_challenges WHERE expires_at <= $1 OR consumed_at IS NOT NULL`,
		now); err != nil {
		@@ -2146,6 +2173,12 @@ func (p *PGRepo) LinkAccount(ctx context.Context, userID, mcUUID, authSource str
		// earlier unfinished migration for the source (so re-running /felis migrate restarts
		// cleanly, invalidating a prior outstanding code) and inserts a fresh 'initiated' row,
		// both under one transaction so the partial unique index never sees two live rows.
		//
		// Starts for one source serialise on a transaction-scoped advisory lock, so racing
		// starts each supersede the one before instead of tripping the unique index. The
		// liveness check rides on the INSERT, after the supersede: a redeem of this source holds
		// the code_issued row until it commits, the supersede waits on that row, and the INSERT's
		// fresh snapshot then sees the source retired (a check before the wait would not).
		func (p *PGRepo) StartMigration(ctx context.Context, id, sourceUserID string, now time.Time) error {
		tx, err := p.db.BeginTx(ctx, nil)
		if err != nil {
		@@ -2153,29 +2186,32 @@ func (p *PGRepo) StartMigration(ctx context.Context, id, sourceUserID string, no
		}
		defer tx.Rollback() //nolint:errcheck // no-op after commit

		// The source must be a live (non-deleted) account; a retired one can never
		// re-initiate a migration.
		var live bool
		if err := tx.QueryRowContext(ctx,
		`SELECT EXISTS(SELECT 1 FROM users WHERE id = $1 AND deleted_at IS NULL)`,
		sourceUserID).Scan(&live); err != nil {
		return err
		}
		if !live {
		return ErrNotFound
		if _, err := tx.ExecContext(ctx,
		`SELECT pg_advisory_xact_lock(hashtext('account-migration:' \|\| $1))`, sourceUserID); err != nil {
		return fmt.Errorf("lock migration source: %w", err)
		}

		if _, err := tx.ExecContext(ctx,
		`DELETE FROM account_migrations WHERE source_user_id = $1 AND state <> 'redeemed'`,
		sourceUserID); err != nil {
		return err
		}
		if _, err := tx.ExecContext(ctx,
		// The source must be a live (non-deleted) account; a retired one can never
		// re-initiate a migration.
		res, err := tx.ExecContext(ctx,
		`INSERT INTO account_migrations (id, source_user_id, state, created_at, updated_at)
		VALUES ($1, $2, 'initiated', $3, $3)`,
		id, sourceUserID, now); err != nil {
		SELECT $1, $2, 'initiated', $3, $3
		WHERE EXISTS (SELECT 1 FROM users WHERE id = $2 AND deleted_at IS NULL)`,
		id, sourceUserID, now)
		if err != nil {
		return err
		}
		n, err := res.RowsAffected()
		if err != nil {
		return err
		}
		if n == 0 {
		return ErrNotFound
		}
		return tx.Commit()
		}

internal/pgint/accounts_test.go

0 → 100644

+518 −0

File added.

Preview size limit exceeded, changes collapsed.

internal/pgint/challenges_test.go

+225 −0

Changes for internal/pgint/challenges_test.go: 225 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -5,6 +5,7 @@ package pgint
		import (
		"context"
		"errors"
		"sync"
		"testing"
		"time"

		@@ -285,3 +286,227 @@ func TestDiscoverableChallengeBounds(t *testing.T) {
		t.Fatalf("begin with the table full = %v, want ErrTooManyPasskeyChallenges", err)
		}
		}

		// ---- enrollment and step-up ceremonies (migration 0007) ---------------------------

		func challengeRows(t *testing.T, userID, purpose string) int {
		t.Helper()
		var n int
		if err := db.QueryRow(`SELECT count(*) FROM webauthn_challenges WHERE user_id = $1 AND purpose = $2`,
		userID, purpose).Scan(&n); err != nil {
		t.Fatalf("count challenges: %v", err)
		}
		return n
		}

		// An enrollment begin supersedes the account's earlier one for the same purpose and
		// sweeps the row its last finish spent, so (user, purpose) holds one row; finish is
		// single-use, dies at expires_at, and never crosses purposes or accounts.
		func TestPasskeyChallengeByUserContract(t *testing.T) {
		ctx := context.Background()
		u := newUser(t, "user", "pk-reg")
		other := newUser(t, "user", "pk-reg-other")
		const reg = "passkey_register"
		t0 := mustNow().Truncate(time.Second)
		begin := func(userID, purpose, session string, expiresAt time.Time) {
		t.Helper()
		if err := repo.CreatePasskeyChallenge(ctx, "reg-"+suffix(t), userID, purpose, []byte(session), expiresAt); err != nil {
		t.Fatalf("CreatePasskeyChallenge(%s): %v", session, err)
		}
		}
		finish := func(userID, purpose string, at time.Time) (string, error) {
		sd, err := repo.ConsumePasskeyChallengeByUser(ctx, userID, purpose, at)
		return string(sd), err
		}

		if _, err := finish(u.ID, reg, t0); !errors.Is(err, api.ErrPasskeyChallengeInvalid) {
		t.Fatalf("finish with no begin = %v, want ErrPasskeyChallengeInvalid", err)
		}
		begin(u.ID, reg, "s1", t0.Add(5*time.Minute))
		begin(other.ID, reg, "o1", t0.Add(5*time.Minute))
		begin(u.ID, reg, "s2", t0.Add(5*time.Minute))
		if n := challengeRows(t, u.ID, reg); n != 1 {
		t.Fatalf("rows after two begins = %d, want 1", n)
		}
		if sd, err := finish(u.ID, reg, t0); err != nil \|\| sd != "s2" {
		t.Fatalf("finish = %q, %v; want s2", sd, err)
		}
		if _, err := finish(u.ID, reg, t0); !errors.Is(err, api.ErrPasskeyChallengeInvalid) {
		t.Fatalf("replayed finish = %v, want ErrPasskeyChallengeInvalid", err)
		}
		if sd, err := finish(other.ID, reg, t0); err != nil \|\| sd != "o1" {
		t.Fatalf("other account's finish = %q, %v; want o1 (a begin never touches another account)", sd, err)
		}

		// The spent row goes with the next begin.
		begin(u.ID, reg, "s3", t0.Add(5*time.Minute))
		if n := challengeRows(t, u.ID, reg); n != 1 {
		t.Fatalf("rows after finish then begin = %d, want 1", n)
		}

		// A step-up ceremony is its own door.
		begin(u.ID, "reauth", "r1", t0.Add(5*time.Minute))
		if n := challengeRows(t, u.ID, reg); n != 1 {
		t.Fatalf("a reauth begin touched the enrollment row: %d rows, want 1", n)
		}
		if sd, err := finish(u.ID, "reauth", t0); err != nil \|\| sd != "r1" {
		t.Fatalf("reauth finish = %q, %v; want r1", sd, err)
		}
		if _, err := finish(u.ID, "reauth", t0); !errors.Is(err, api.ErrPasskeyChallengeInvalid) {
		t.Fatalf("reauth replay = %v, want ErrPasskeyChallengeInvalid", err)
		}

		// Expiry: at expires_at the ceremony is dead; a second earlier it still redeems.
		if _, err := finish(u.ID, reg, t0.Add(5*time.Minute)); !errors.Is(err, api.ErrPasskeyChallengeInvalid) {
		t.Fatalf("finish at expires_at = %v, want ErrPasskeyChallengeInvalid", err)
		}
		if sd, err := finish(u.ID, reg, t0.Add(5*time.Minute-time.Second)); err != nil \|\| sd != "s3" {
		t.Fatalf("finish a second before expiry = %q, %v; want s3", sd, err)
		}
		}

		// Begins that race for one (user, purpose) leave exactly one row, and finishes that
		// race for one ceremony redeem it once.
		func TestPasskeyChallengeByUserUnderConcurrency(t *testing.T) {
		ctx := context.Background()
		const reg = "passkey_register"
		now := mustNow()
		for round := 0; round < 5; round++ {
		u := newUser(t, "user", "pk-race")
		var wg sync.WaitGroup
		errs := make([]error, 8)
		for i := range errs {
		wg.Add(1)
		go func(i int) {
		defer wg.Done()
		errs[i] = repo.CreatePasskeyChallenge(ctx, "race-"+suffix(t), u.ID, reg, []byte("s"), now.Add(5*time.Minute))
		}(i)
		}
		wg.Wait()
		for i, err := range errs {
		if err != nil {
		t.Fatalf("round %d: begin %d: %v", round, i, err)
		}
		}
		if n := challengeRows(t, u.ID, reg); n != 1 {
		t.Fatalf("round %d: rows after 8 racing begins = %d, want 1", round, n)
		}

		won, lost := 0, 0
		for i := range errs {
		wg.Add(1)
		go func(i int) {
		defer wg.Done()
		_, errs[i] = repo.ConsumePasskeyChallengeByUser(ctx, u.ID, reg, now)
		}(i)
		}
		wg.Wait()
		for i, err := range errs {
		switch {
		case err == nil:
		won++
		case errors.Is(err, api.ErrPasskeyChallengeInvalid):
		lost++
		default:
		t.Fatalf("round %d: finish %d: %v", round, i, err)
		}
		}
		if won != 1 \|\| lost != 7 {
		t.Fatalf("round %d: racing finishes: %d redeemed, %d refused; want 1, 7", round, won, lost)
		}
		}
		}

		// Racing begins from one source still stop at 32 live challenges, in both login
		// stores: the count and the insert are one decision.
		func TestLoginChallengeSourceBoundUnderConcurrency(t *testing.T) {
		ctx := context.Background()
		now := mustNow()
		race := func(t *testing.T, begin func(i int) error) (ok, refused int) {
		t.Helper()
		var wg sync.WaitGroup
		errs := make([]error, 64)
		for i := range errs {
		wg.Add(1)
		go func(i int) {
		defer wg.Done()
		errs[i] = begin(i)
		}(i)
		}
		wg.Wait()
		for i, err := range errs {
		switch {
		case err == nil:
		ok++
		case errors.Is(err, api.ErrTooManyPasskeyChallenges):
		refused++
		default:
		t.Fatalf("begin %d: %v", i, err)
		}
		}
		return ok, refused
		}

		t.Run("discoverable", func(t *testing.T) {
		source := "2001:db8:" + suffix(t)[:4] + "::/48"
		t.Cleanup(func() {
		db.Exec(`DELETE FROM webauthn_discoverable_challenges WHERE source = $1`, source) //nolint:errcheck
		})
		ok, refused := race(t, func(int) error {
		return repo.CreateDiscoverableChallenge(ctx, "drace-"+suffix(t), source, []byte("s"), now, now.Add(5*time.Minute))
		})
		var stored int
		if err := db.QueryRow(`SELECT count(*) FROM webauthn_discoverable_challenges WHERE source = $1`, source).Scan(&stored); err != nil {
		t.Fatalf("count: %v", err)
		}
		if ok != 32 \|\| refused != 32 \|\| stored != 32 {
		t.Fatalf("64 racing begins: %d stored, %d refused, %d rows; want 32, 32, 32", ok, refused, stored)
		}
		})

		t.Run("email-first", func(t *testing.T) {
		u := newUser(t, "user", "pk-login-race")
		source := "203.0.113." + suffix(t)
		ok, refused := race(t, func(int) error {
		return repo.AddPasskeyLoginChallenge(ctx, "lrace-"+suffix(t), u.ID, "passkey_login", source, "c-"+suffix(t), []byte("s"), now, now.Add(5*time.Minute))
		})
		var stored int
		if err := db.QueryRow(`SELECT count(*) FROM webauthn_challenges WHERE source = $1`, source).Scan(&stored); err != nil {
		t.Fatalf("count: %v", err)
		}
		if ok != 32 \|\| refused != 32 \|\| stored != 32 {
		t.Fatalf("64 racing begins: %d stored, %d refused, %d rows; want 32, 32, 32", ok, refused, stored)
		}
		})

		t.Run("discoverable finish", func(t *testing.T) {
		id := "dfin-" + suffix(t)
		if err := repo.CreateDiscoverableChallenge(ctx, id, "pgint-fin-"+suffix(t), []byte("s"), now, now.Add(5*time.Minute)); err != nil {
		t.Fatalf("begin: %v", err)
		}
		var wg sync.WaitGroup
		errs := make([]error, 8)
		for i := range errs {
		wg.Add(1)
		go func(i int) {
		defer wg.Done()
		_, errs[i] = repo.ConsumeDiscoverableChallenge(ctx, id, now)
		}(i)
		}
		wg.Wait()
		won, lost := 0, 0
		for i, err := range errs {
		switch {
		case err == nil:
		won++
		case errors.Is(err, api.ErrPasskeyChallengeInvalid):
		lost++
		default:
		t.Fatalf("finish %d: %v", i, err)
		}
		}
		if won != 1 \|\| lost != 7 {
		t.Fatalf("racing finishes: %d redeemed, %d refused; want 1, 7", won, lost)
		}
		})
		}