etcdctl/ctlv3: auth: wildcard terminated paths specify ranges.

 Implement wildcard '*' for get and del.
 When '*' is the last symbol in a key,
 the key is converted to a prefix range.
 The wildcard may only appear as the last
 symbol in a key.

 Also: fix several fencepost bugs fixed in
 the range sorting/merge code, including one
 incorrect test in auth/range_perm_cache_test.go

 Fixes etcd-io#6359

auth/range_perm_cache.go

@@ -22,22 +22,53 @@ import (
 	"github.com/coreos/etcd/mvcc/backend"
 )
 
-// isSubset returns true if a is a subset of b
+// isSubset returns true if a is a subset of b.
+// If a is a prefix of b, then a is a subset of b.
+// Given intervals [a1,a2) and [b1,b2), is
+// the a interval a subset of b?
 func isSubset(a, b *rangePerm) bool {
 	switch {
 	case len(a.end) == 0 && len(b.end) == 0:
 		// a, b are both keys
 		return bytes.Equal(a.begin, b.begin)
 	case len(b.end) == 0:
-		// b is a key, a is a range
+		// b is a key, a is a range (even a prefix range has infinite membership > 1)
 		return false
 	case len(a.end) == 0:
-		return 0 <= bytes.Compare(a.begin, b.begin) && bytes.Compare(a.begin, b.end) <= 0
+		// a is a key, b is a range. need b1 <= a1 and a1 < b2
+		return bytes.Compare(b.begin, a.begin) <= 0 && bytes.Compare(a.begin, b.end) < 0
 	default:
-		return 0 <= bytes.Compare(a.begin, b.begin) && bytes.Compare(a.end, b.end) <= 0
+		// both are ranges. need b1 <= a1 and a2 <= b2
+		return bytes.Compare(b.begin, a.begin) <= 0 && bytes.Compare(a.end, b.end) <= 0
 	}
 }
 
+func rangeIsPrefix(a *rangePerm) bool {
+	if len(a.end) == 0 {
+		return true
+	}
+	lenbeg := len(a.begin)
+	lenend := len(a.end)
+	x := a.begin[lenbeg-1]
+	y := a.end[lenend-1]
+	if lenbeg == lenend {
+		if x+1 == y {
+			return true
+		}
+		return false
+	}
+	// check for overflow
+	if lenend != 1+lenbeg {
+		return false
+	}
+	// INVAR: lenend is one greater than lenbeg, might have overflowed
+	if x+1 == 0 && y == 0 {
+		// yep, overflowed.
+		return true
+	}
+	return false
+}
+
 func isRangeEqual(a, b *rangePerm) bool {
 	return bytes.Equal(a.begin, b.begin) && bytes.Equal(a.end, b.end)
 }
@@ -88,12 +119,18 @@ func mergeRangePerms(perms []*rangePerm) []*rangePerm {
 	i := 0
 	for i < len(perms) {
 		begin, next := i, i
-		for next+1 < len(perms) && bytes.Compare(perms[next].end, perms[next+1].begin) != -1 {
+		for next+1 < len(perms) && bytes.Compare(perms[next].end, perms[next+1].begin) >= 0 {
 			next++
 		}
-
-		merged = append(merged, &rangePerm{begin: perms[begin].begin, end: perms[next].end})
-
+		// don't merge ["a", "b") with ["b", ""), because perms[next+1].end is empty.
+		if next != begin && len(perms[next].end) > 0 {
+			merged = append(merged, &rangePerm{begin: perms[begin].begin, end: perms[next].end})
+		} else {
+			merged = append(merged, perms[begin])
+			if next != begin {
+				merged = append(merged, perms[next])
+			}
+		}
 		i = next + 1
 	}
 
@@ -151,8 +188,11 @@ func checkKeyPerm(cachedPerms *unifiedRangePermissions, key, rangeEnd []byte, pe
 	}
 
 	requiredPerm := &rangePerm{begin: key, end: rangeEnd}
+	expandPrefixToRange(requiredPerm, '*', true)
 
 	for _, perm := range tocheck {
+		expandPrefixToRange(perm, '*', true)
+
 		if isSubset(requiredPerm, perm) {
 			return true
 		}
@@ -161,6 +201,43 @@ func checkKeyPerm(cachedPerms *unifiedRangePermissions, key, rangeEnd []byte, pe
 	return false
 }
 
+// expandPrefixToRange converts a single key
+// with wildcard as its last character into a range.
+// For example, we will convert a range like ['/home/user/*', '')
+// into ['/home/user/', '/home/user0').
+// Returns true if conversion/expansion took place.
+// If removeWild is true, then we will remove the wildcard byte
+// before creating the range.
+func expandPrefixToRange(r *rangePerm, wildcard byte, removeWild bool) bool {
+	lenkey := len(r.begin)
+	lenend := len(r.end)
+	if lenend == 0 && lenkey > 0 && r.begin[lenkey-1] == wildcard {
+		if lenkey == 1 {
+			// request for all keys
+			r.begin = []byte{0}
+			r.end = []byte{0}
+			return true
+		}
+		// we have a wildcard query, with keylen >= 2. Fix the begin and end:
+		if removeWild {
+			r.begin = r.begin[:lenkey-1] // remove the wildcard
+		}
+		lenkey = len(r.begin)
+
+		// setting RangeEnd one bit higher makes a prefix query
+		r.end = make([]byte, lenkey)
+		copy(r.end, r.begin)
+		r.end[lenkey-1]++
+		// check for overflow
+		if r.end[lenkey-1] == 0 {
+			// yep, overflowed.
+			r.end = append(r.end, 0)
+		}
+		return true
+	}
+	return false
+}
+
 func (as *authStore) isRangeOpPermitted(tx backend.BatchTx, userName string, key, rangeEnd []byte, permtyp authpb.Permission_Type) bool {
 	// assumption: tx is Lock()ed
 	_, ok := as.rangePermCache[userName]

auth/range_perm_cache_test.go

@@ -106,7 +106,7 @@ func TestGetMergedPerms(t *testing.T) {
 		},
 		{
 			[]*rangePerm{{[]byte("a"), []byte("")}, {[]byte("b"), []byte("c")}, {[]byte("b"), []byte("")}, {[]byte("c"), []byte("")}, {[]byte("d"), []byte("")}},
-			[]*rangePerm{{[]byte("a"), []byte("")}, {[]byte("b"), []byte("c")}, {[]byte("d"), []byte("")}},
+			[]*rangePerm{{[]byte("a"), []byte("")}, {[]byte("b"), []byte("c")}, {[]byte("c"), []byte("")}, {[]byte("d"), []byte("")}},
 		},
 		// duplicate ranges
 		{
@@ -123,7 +123,7 @@ func TestGetMergedPerms(t *testing.T) {
 	for i, tt := range tests {
 		result := mergeRangePerms(tt.params)
 		if !isPermsEqual(result, tt.want) {
-			t.Errorf("#%d: result=%q, want=%q", i, result, tt.want)
+			t.Fatalf("#%d: result=%q, want=%q", i, result, tt.want)
 		}
 	}
 }

etcdserver/apply_auth.go

@@ -75,10 +75,66 @@ func (aa *authApplierV3) Range(txnID int64, r *pb.RangeRequest) (*pb.RangeRespon
 	if err := aa.as.IsRangePermitted(&aa.authInfo, r.Key, r.RangeEnd); err != nil {
 		return nil, err
 	}
+	AllowWildcardGets(r)
 	return aa.applierV3.Range(txnID, r)
 }
 
+// AllowWildcardGets enables get '*' to return all keys,
+// and get '/home/user/*' to return all keys with
+// the prefix '/home/user/'.
+func AllowWildcardGets(r *pb.RangeRequest) {
+	lenkey := len(r.Key)
+	lenend := len(r.RangeEnd)
+	if lenend == 0 && lenkey > 0 && r.Key[lenkey-1] == '*' {
+		if lenkey == 1 {
+			// request for all keys
+			r.Key = []byte{0}
+			r.RangeEnd = []byte{0}
+			return
+		}
+		// we have a wildcard query, with keylen >= 2. Fix the begin and end:
+		r.Key = r.Key[:lenkey-1] // remove the '*'
+		// setting RangeEnd one bit higher makes a prefix query
+		r.RangeEnd = make([]byte, lenkey-1)
+		copy(r.RangeEnd, r.Key)
+		r.RangeEnd[lenkey-2]++
+		// check for overflow
+		if r.RangeEnd[lenkey-2] == 0 {
+			// yep, overflowed.
+			r.RangeEnd = append(r.RangeEnd, 0)
+		}
+	}
+}
+
+// AllowWildcardDeletes enables del '*' to delete all keys,
+// and del '/home/user/*' to delete all keys with
+// the prefix '/home/user/'.
+func AllowWildcardDeletes(r *pb.DeleteRangeRequest) {
+	lenkey := len(r.Key)
+	lenend := len(r.RangeEnd)
+	if lenend == 0 && lenkey > 0 && r.Key[lenkey-1] == '*' {
+		if lenkey == 1 {
+			// request for all keys
+			r.Key = []byte{0}
+			r.RangeEnd = []byte{0}
+			return
+		}
+		// we have a wildcard query, with keylen >= 2. Fix the begin and end:
+		r.Key = r.Key[:lenkey-1] // remove the '*'
+		// setting RangeEnd one bit higher makes a prefix query
+		r.RangeEnd = make([]byte, lenkey-1)
+		copy(r.RangeEnd, r.Key)
+		r.RangeEnd[lenkey-2]++
+		// check for overflow
+		if r.RangeEnd[lenkey-2] == 0 {
+			// yep, overflowed.
+			r.RangeEnd = append(r.RangeEnd, 0)
+		}
+	}
+}
+
 func (aa *authApplierV3) DeleteRange(txnID int64, r *pb.DeleteRangeRequest) (*pb.DeleteRangeResponse, error) {
+	AllowWildcardDeletes(r)
 	if err := aa.as.IsDeleteRangePermitted(&aa.authInfo, r.Key, r.RangeEnd); err != nil {
 		return nil, err
 	}