fixed

broker/auth.go

@@ -14,6 +14,10 @@ const (
 )
 
 func (c *client) CheckTopicAuth(typ int, topic string) bool {
+	if c.typ != CLIENT || !c.broker.pluginAuthHTTP {
+		return true
+	}
+
 	if strings.HasPrefix(topic, "$SYS/broker/connection/clients/") {
 		return true
 	}
@@ -22,9 +26,6 @@ func (c *client) CheckTopicAuth(typ int, topic string) bool {
 		topic = strings.TrimPrefix(topic, "$queue/")
 	}
 
-	if c.typ != CLIENT || !c.broker.pluginAuthHTTP {
-		return true
-	}
 	access := "sub"
 	switch typ {
 	case 1:
@@ -36,3 +37,15 @@ func (c *client) CheckTopicAuth(typ int, topic string) bool {
 	return authhttp.CheckACL(username, access, topic)
 
 }
+
+func (b *Broker) CheckConnectAuth(clientID, username, password string) bool {
+	if b.pluginAuthHTTP {
+		if clientID == "" || username == "" {
+			return false
+		}
+		return authhttp.CheckAuth(clientID, username, password)
+	}
+
+	return false
+
+}

broker/broker.go

@@ -50,8 +50,6 @@ type Broker struct {
 	sessionMgr     *sessions.Manager
 	pluginAuthHTTP bool
 	pluginKafka    bool
-
-	// messagePool []chan *Message
 }
 
 func newMessagePool() []chan *Message {
@@ -329,10 +327,19 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) {
 	log.Info("read connect from ", zap.String("clientID", msg.ClientIdentifier))
 
 	connack := packets.NewControlPacket(packets.Connack).(*packets.ConnackPacket)
-	connack.ReturnCode = packets.Accepted
 	connack.SessionPresent = msg.CleanSession
+	connack.ReturnCode = msg.Validate()
 
-	if b.pluginAuthHTTP == true && authhttp.CheckAuth(string(msg.ClientIdentifier), string(msg.Username), string(msg.Password)) {
+	if connack.ReturnCode != packets.Accepted {
+		err = connack.Write(conn)
+		if err != nil {
+			log.Error("send connack error, ", zap.Error(err), zap.String("clientID", msg.ClientIdentifier))
+			return
+		}
+	}
+
+	if typ == CLIENT && b.CheckConnectAuth(string(msg.ClientIdentifier), string(msg.Username), string(msg.Password)) {
+		connack.ReturnCode = packets.ErrRefusedNotAuthorised
 		err = connack.Write(conn)
 		if err != nil {
 			log.Error("send connack error, ", zap.Error(err), zap.String("clientID", msg.ClientIdentifier))

go.mod

@@ -11,6 +11,7 @@ require (
 	github.com/gin-gonic/gin v1.4.0
 	github.com/go-ole/go-ole v1.2.4 // indirect
 	github.com/kr/pretty v0.1.0 // indirect
+	github.com/patrickmn/go-cache v2.1.0+incompatible
 	github.com/pkg/errors v0.8.1 // indirect
 	github.com/satori/go.uuid v1.2.0
 	github.com/segmentio/fasthash v0.0.0-20180216231524-a72b379d632e

go.sum

@@ -49,6 +49,8 @@ github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/reflect2 v1.0.1 h1:9f412s+6RmYXLWZSEzVVgPGK7C2PphHj5RJrvfx9AWI=
 github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
+github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=
+github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
 github.com/pierrec/lz4 v0.0.0-20190327172049-315a67e90e41 h1:GeinFsrjWz97fAxVUEd748aV0cYL+I6k44gFJTCVvpU=
 github.com/pierrec/lz4 v0.0.0-20190327172049-315a67e90e41/go.mod h1:3/3N9NVKO0jef7pBehbT1qWhCMrIgbYNnFAZCqQ5LRc=
 github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=

plugins/authhttp/authhttp.go

@@ -58,6 +58,16 @@ func Init() {
 
 //CheckAuth check mqtt connect
 func CheckAuth(clientID, username, password string) bool {
+	action := "connect"
+	{
+		aCache := checkCache(action, clientID, username, password, "")
+		if aCache != nil {
+			if aCache.password == password && aCache.username == username && aCache.action == action {
+				return true
+			}
+		}
+	}
+
 	data := url.Values{}
 	data.Add("username", username)
 	data.Add("clientid", clientID)
@@ -79,6 +89,7 @@ func CheckAuth(clientID, username, password string) bool {
 
 	defer resp.Body.Close()
 	if resp.StatusCode == http.StatusOK {
+		addCache(action, clientID, username, password, "")
 		return true
 	}
 	return false
@@ -86,6 +97,16 @@ func CheckAuth(clientID, username, password string) bool {
 
 //CheckSuper check mqtt connect
 func CheckSuper(clientID, username, password string) bool {
+	action := "connect"
+	{
+		aCache := checkCache(action, clientID, username, password, "")
+		if aCache != nil {
+			if aCache.password == password && aCache.username == username && aCache.action == action {
+				return true
+			}
+		}
+	}
+
 	data := url.Values{}
 	data.Add("username", username)
 	data.Add("clientid", clientID)
@@ -114,6 +135,16 @@ func CheckSuper(clientID, username, password string) bool {
 
 //CheckACL check mqtt connect
 func CheckACL(username, access, topic string) bool {
+	action := access
+	{
+		aCache := checkCache(action, "", username, "", topic)
+		if aCache != nil {
+			if aCache.topic == topic && aCache.action == action {
+				return true
+			}
+		}
+	}
+
 	req, err := http.NewRequest("GET", config.ACLURL, nil)
 	if err != nil {
 		log.Error("get acl: ", zap.Error(err))
@@ -135,6 +166,7 @@ func CheckACL(username, access, topic string) bool {
 
 	defer resp.Body.Close()
 	if resp.StatusCode == http.StatusOK {
+		addCache(action, "", username, "", topic)
 		return true
 	}
 	return false

plugins/authhttp/cache.go

@@ -0,0 +1,32 @@
+package authhttp
+
+import (
+	"time"
+
+	"github.com/patrickmn/go-cache"
+)
+
+type authCache struct {
+	action   string
+	username string
+	clientID string
+	password string
+	topic    string
+}
+
+var (
+	// cache = make(map[string]authCache)
+	c = cache.New(5*time.Minute, 10*time.Minute)
+)
+
+func checkCache(action, clientID, username, password, topic string) *authCache {
+	authc, found := c.Get(username)
+	if found {
+		return authc.(*authCache)
+	}
+	return nil
+}
+
+func addCache(action, clientID, username, password, topic string) {
+	c.Set(username, &authCache{action: action, username: username, clientID: clientID, password: password, topic: topic}, cache.DefaultExpiration)
+}

vendor/github.com/patrickmn/go-cache/CONTRIBUTORS

@@ -0,0 +1,9 @@
+This is a list of people who have contributed code to go-cache. They, or their
+employers, are the copyright holders of the contributed code. Contributed code
+is subject to the license restrictions listed in LICENSE (as they were when the
+code was contributed.)
+
+Dustin Sallings <dustin@spy.net>
+Jason Mooberry <jasonmoo@me.com>
+Sergey Shepelev <temotor@gmail.com>
+Alex Edwards <ajmedwards@gmail.com>

vendor/github.com/patrickmn/go-cache/LICENSE

@@ -0,0 +1,19 @@
+Copyright (c) 2012-2017 Patrick Mylund Nielsen and the go-cache contributors
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.

vendor/github.com/patrickmn/go-cache/README.md

@@ -0,0 +1,83 @@
+# go-cache
+
+go-cache is an in-memory key:value store/cache similar to memcached that is
+suitable for applications running on a single machine. Its major advantage is
+that, being essentially a thread-safe `map[string]interface{}` with expiration
+times, it doesn't need to serialize or transmit its contents over the network.
+
+Any object can be stored, for a given duration or forever, and the cache can be
+safely used by multiple goroutines.
+
+Although go-cache isn't meant to be used as a persistent datastore, the entire
+cache can be saved to and loaded from a file (using `c.Items()` to retrieve the
+items map to serialize, and `NewFrom()` to create a cache from a deserialized
+one) to recover from downtime quickly. (See the docs for `NewFrom()` for caveats.)
+
+### Installation
+
+`go get github.com/patrickmn/go-cache`
+
+### Usage
+
+```go
+import (
+	"fmt"
+	"github.com/patrickmn/go-cache"
+	"time"
+)
+
+func main() {
+	// Create a cache with a default expiration time of 5 minutes, and which
+	// purges expired items every 10 minutes
+	c := cache.New(5*time.Minute, 10*time.Minute)
+
+	// Set the value of the key "foo" to "bar", with the default expiration time
+	c.Set("foo", "bar", cache.DefaultExpiration)
+
+	// Set the value of the key "baz" to 42, with no expiration time
+	// (the item won't be removed until it is re-set, or removed using
+	// c.Delete("baz")
+	c.Set("baz", 42, cache.NoExpiration)
+
+	// Get the string associated with the key "foo" from the cache
+	foo, found := c.Get("foo")
+	if found {
+		fmt.Println(foo)
+	}
+
+	// Since Go is statically typed, and cache values can be anything, type
+	// assertion is needed when values are being passed to functions that don't
+	// take arbitrary types, (i.e. interface{}). The simplest way to do this for
+	// values which will only be used once--e.g. for passing to another
+	// function--is:
+	foo, found := c.Get("foo")
+	if found {
+		MyFunction(foo.(string))
+	}
+
+	// This gets tedious if the value is used several times in the same function.
+	// You might do either of the following instead:
+	if x, found := c.Get("foo"); found {
+		foo := x.(string)
+		// ...
+	}
+	// or
+	var foo string
+	if x, found := c.Get("foo"); found {
+		foo = x.(string)
+	}
+	// ...
+	// foo can then be passed around freely as a string
+
+	// Want performance? Store pointers!
+	c.Set("foo", &MyStruct, cache.DefaultExpiration)
+	if x, found := c.Get("foo"); found {
+		foo := x.(*MyStruct)
+			// ...
+	}
+}
+```
+
+### Reference
+
+`godoc` or [http://godoc.org/github.com/patrickmn/go-cache](http://godoc.org/github.com/patrickmn/go-cache)

vendor/github.com/patrickmn/go-cache/sharded.go

@@ -0,0 +1,192 @@
+package cache
+
+import (
+	"crypto/rand"
+	"math"
+	"math/big"
+	insecurerand "math/rand"
+	"os"
+	"runtime"
+	"time"
+)
+
+// This is an experimental and unexported (for now) attempt at making a cache
+// with better algorithmic complexity than the standard one, namely by
+// preventing write locks of the entire cache when an item is added. As of the
+// time of writing, the overhead of selecting buckets results in cache
+// operations being about twice as slow as for the standard cache with small
+// total cache sizes, and faster for larger ones.
+//
+// See cache_test.go for a few benchmarks.
+
+type unexportedShardedCache struct {
+	*shardedCache
+}
+
+type shardedCache struct {
+	seed    uint32
+	m       uint32
+	cs      []*cache
+	janitor *shardedJanitor
+}
+
+// djb2 with better shuffling. 5x faster than FNV with the hash.Hash overhead.
+func djb33(seed uint32, k string) uint32 {
+	var (
+		l = uint32(len(k))
+		d = 5381 + seed + l
+		i = uint32(0)
+	)
+	// Why is all this 5x faster than a for loop?
+	if l >= 4 {
+		for i < l-4 {
+			d = (d * 33) ^ uint32(k[i])
+			d = (d * 33) ^ uint32(k[i+1])
+			d = (d * 33) ^ uint32(k[i+2])
+			d = (d * 33) ^ uint32(k[i+3])
+			i += 4
+		}
+	}
+	switch l - i {
+	case 1:
+	case 2:
+		d = (d * 33) ^ uint32(k[i])
+	case 3:
+		d = (d * 33) ^ uint32(k[i])
+		d = (d * 33) ^ uint32(k[i+1])
+	case 4:
+		d = (d * 33) ^ uint32(k[i])
+		d = (d * 33) ^ uint32(k[i+1])
+		d = (d * 33) ^ uint32(k[i+2])
+	}
+	return d ^ (d >> 16)
+}
+
+func (sc *shardedCache) bucket(k string) *cache {
+	return sc.cs[djb33(sc.seed, k)%sc.m]
+}
+
+func (sc *shardedCache) Set(k string, x interface{}, d time.Duration) {
+	sc.bucket(k).Set(k, x, d)
+}
+
+func (sc *shardedCache) Add(k string, x interface{}, d time.Duration) error {
+	return sc.bucket(k).Add(k, x, d)
+}
+
+func (sc *shardedCache) Replace(k string, x interface{}, d time.Duration) error {
+	return sc.bucket(k).Replace(k, x, d)
+}
+
+func (sc *shardedCache) Get(k string) (interface{}, bool) {
+	return sc.bucket(k).Get(k)
+}
+
+func (sc *shardedCache) Increment(k string, n int64) error {
+	return sc.bucket(k).Increment(k, n)
+}
+
+func (sc *shardedCache) IncrementFloat(k string, n float64) error {
+	return sc.bucket(k).IncrementFloat(k, n)
+}
+
+func (sc *shardedCache) Decrement(k string, n int64) error {
+	return sc.bucket(k).Decrement(k, n)
+}
+
+func (sc *shardedCache) Delete(k string) {
+	sc.bucket(k).Delete(k)
+}
+
+func (sc *shardedCache) DeleteExpired() {
+	for _, v := range sc.cs {
+		v.DeleteExpired()
+	}
+}
+
+// Returns the items in the cache. This may include items that have expired,
+// but have not yet been cleaned up. If this is significant, the Expiration
+// fields of the items should be checked. Note that explicit synchronization
+// is needed to use a cache and its corresponding Items() return values at
+// the same time, as the maps are shared.
+func (sc *shardedCache) Items() []map[string]Item {
+	res := make([]map[string]Item, len(sc.cs))
+	for i, v := range sc.cs {
+		res[i] = v.Items()
+	}
+	return res
+}
+
+func (sc *shardedCache) Flush() {
+	for _, v := range sc.cs {
+		v.Flush()
+	}
+}
+
+type shardedJanitor struct {
+	Interval time.Duration
+	stop     chan bool
+}
+
+func (j *shardedJanitor) Run(sc *shardedCache) {
+	j.stop = make(chan bool)
+	tick := time.Tick(j.Interval)
+	for {
+		select {
+		case <-tick:
+			sc.DeleteExpired()
+		case <-j.stop:
+			return
+		}
+	}
+}
+
+func stopShardedJanitor(sc *unexportedShardedCache) {
+	sc.janitor.stop <- true
+}
+
+func runShardedJanitor(sc *shardedCache, ci time.Duration) {
+	j := &shardedJanitor{
+		Interval: ci,
+	}
+	sc.janitor = j
+	go j.Run(sc)
+}
+
+func newShardedCache(n int, de time.Duration) *shardedCache {
+	max := big.NewInt(0).SetUint64(uint64(math.MaxUint32))
+	rnd, err := rand.Int(rand.Reader, max)
+	var seed uint32
+	if err != nil {
+		os.Stderr.Write([]byte("WARNING: go-cache's newShardedCache failed to read from the system CSPRNG (/dev/urandom or equivalent.) Your system's security may be compromised. Continuing with an insecure seed.\n"))
+		seed = insecurerand.Uint32()
+	} else {
+		seed = uint32(rnd.Uint64())
+	}
+	sc := &shardedCache{
+		seed: seed,
+		m:    uint32(n),
+		cs:   make([]*cache, n),
+	}
+	for i := 0; i < n; i++ {
+		c := &cache{
+			defaultExpiration: de,
+			items:             map[string]Item{},
+		}
+		sc.cs[i] = c
+	}
+	return sc
+}
+
+func unexportedNewSharded(defaultExpiration, cleanupInterval time.Duration, shards int) *unexportedShardedCache {
+	if defaultExpiration == 0 {
+		defaultExpiration = -1
+	}
+	sc := newShardedCache(shards, defaultExpiration)
+	SC := &unexportedShardedCache{sc}
+	if cleanupInterval > 0 {
+		runShardedJanitor(sc, cleanupInterval)
+		runtime.SetFinalizer(SC, stopShardedJanitor)
+	}
+	return SC
+}

vendor/modules.txt

@@ -43,6 +43,8 @@ github.com/mattn/go-isatty
 github.com/modern-go/concurrent
 # github.com/modern-go/reflect2 v1.0.1
 github.com/modern-go/reflect2
+# github.com/patrickmn/go-cache v2.1.0+incompatible
+github.com/patrickmn/go-cache
 # github.com/pierrec/lz4 v0.0.0-20190327172049-315a67e90e41
 github.com/pierrec/lz4
 github.com/pierrec/lz4/internal/xxh32