Merge pull request #7664 from gyuho/safe-revision-access

auth: use atomic access to 'authStore.revision'

auth/store.go

@@ -21,6 +21,7 @@ import (
 	"sort"
 	"strings"
 	"sync"
+	"sync/atomic"
 
 	"github.com/coreos/etcd/auth/authpb"
 	pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
@@ -174,14 +175,15 @@ type TokenProvider interface {
 }
 
 type authStore struct {
+	// atomic operations; need 64-bit align, or 32-bit tests will crash
+	revision uint64
+
 	be        backend.Backend
 	enabled   bool
 	enabledMu sync.RWMutex
 
 	rangePermCache map[string]*unifiedRangePermissions // username -> unifiedRangePermissions
 
-	revision uint64
-
 	tokenProvider TokenProvider
 }
 
@@ -216,7 +218,7 @@ func (as *authStore) AuthEnable() error {
 
 	as.rangePermCache = make(map[string]*unifiedRangePermissions)
 
-	as.revision = getRevision(tx)
+	as.setRevision(getRevision(tx))
 
 	plog.Noticef("Authentication enabled")
 
@@ -270,7 +272,7 @@ func (as *authStore) Authenticate(ctx context.Context, username, password string
 	// Password checking is already performed in the API layer, so we don't need to check for now.
 	// Staleness of password can be detected with OCC in the API layer, too.
 
-	token, err := as.tokenProvider.assign(ctx, username, as.revision)
+	token, err := as.tokenProvider.assign(ctx, username, as.Revision())
 	if err != nil {
 		return nil, err
 	}
@@ -309,7 +311,7 @@ func (as *authStore) Recover(be backend.Backend) {
 		}
 	}
 
-	as.revision = getRevision(tx)
+	as.setRevision(getRevision(tx))
 
 	tx.Unlock()
 
@@ -658,7 +660,7 @@ func (as *authStore) RoleAdd(r *pb.AuthRoleAddRequest) (*pb.AuthRoleAddResponse,
 }
 
 func (as *authStore) authInfoFromToken(ctx context.Context, token string) (*AuthInfo, bool) {
-	return as.tokenProvider.info(ctx, token, as.revision)
+	return as.tokenProvider.info(ctx, token, as.Revision())
 }
 
 type permSlice []*authpb.Permission
@@ -728,7 +730,7 @@ func (as *authStore) isOpPermitted(userName string, revision uint64, key, rangeE
 		return ErrUserEmpty
 	}
 
-	if revision < as.revision {
+	if revision < as.Revision() {
 		return ErrAuthOldRevision
 	}
 
@@ -919,7 +921,7 @@ func NewAuthStore(be backend.Backend, tp TokenProvider) *authStore {
 		as.tokenProvider.enable()
 	}
 
-	if as.revision == 0 {
+	if as.Revision() == 0 {
 		as.commitRevision(tx)
 	}
 
@@ -939,9 +941,9 @@ func hasRootRole(u *authpb.User) bool {
 }
 
 func (as *authStore) commitRevision(tx backend.BatchTx) {
-	as.revision++
+	atomic.AddUint64(&as.revision, 1)
 	revBytes := make([]byte, revBytesLen)
-	binary.BigEndian.PutUint64(revBytes, as.revision)
+	binary.BigEndian.PutUint64(revBytes, as.Revision())
 	tx.UnsafePut(authBucketName, revisionKey, revBytes)
 }
 
@@ -955,8 +957,12 @@ func getRevision(tx backend.BatchTx) uint64 {
 	return binary.BigEndian.Uint64(vs[0])
 }
 
+func (as *authStore) setRevision(rev uint64) {
+	atomic.StoreUint64(&as.revision, rev)
+}
+
 func (as *authStore) Revision() uint64 {
-	return as.revision
+	return atomic.LoadUint64(&as.revision)
 }
 
 func (as *authStore) AuthInfoFromTLS(ctx context.Context) *AuthInfo {

auth/store_test.go

@@ -506,6 +506,28 @@ func TestAuthDisable(t *testing.T) {
 	}
 }
 
+// TestAuthRevisionRace ensures that access to authStore.revision is thread-safe.
+func TestAuthInfoFromCtxRace(t *testing.T) {
+	b, tPath := backend.NewDefaultTmpBackend()
+	defer os.Remove(tPath)
+
+	tp, err := NewTokenProvider("simple", dummyIndexWaiter)
+	if err != nil {
+		t.Fatal(err)
+	}
+	as := NewAuthStore(b, tp)
+	defer as.Close()
+
+	donec := make(chan struct{})
+	go func() {
+		defer close(donec)
+		ctx := metadata.NewContext(context.Background(), metadata.New(map[string]string{"token": "test"}))
+		as.AuthInfoFromCtx(ctx)
+	}()
+	as.UserAdd(&pb.AuthUserAddRequest{Name: "test"})
+	<-donec
+}
+
 func TestIsAdminPermitted(t *testing.T) {
 	as, tearDown := setupAuthStore(t)
 	defer tearDown(t)