Merge branch 'main' into twilio

server/errors.go

@@ -106,8 +106,10 @@ var (
 	errHTTPBadRequestNotAPaidUser                    = &errHTTP{40027, http.StatusBadRequest, "invalid request: not a paid user", "", nil}
 	errHTTPBadRequestBillingRequestInvalid           = &errHTTP{40028, http.StatusBadRequest, "invalid request: not a valid billing request", "", nil}
 	errHTTPBadRequestBillingSubscriptionExists       = &errHTTP{40029, http.StatusBadRequest, "invalid request: billing subscription already exists", "", nil}
-	errHTTPBadRequestTwilioDisabled                  = &errHTTP{40030, http.StatusBadRequest, "invalid request: Calling is disabled", "https://ntfy.sh/docs/publish/#phone-calls", nil}
-	errHTTPBadRequestPhoneNumberInvalid              = &errHTTP{40031, http.StatusBadRequest, "invalid request: phone number invalid", "https://ntfy.sh/docs/publish/#phone-calls", nil}
+	errHTTPBadRequestTierInvalid                     = &errHTTP{40030, http.StatusBadRequest, "invalid request: tier does not exist", "", nil}
+	errHTTPBadRequestUserNotFound                    = &errHTTP{40031, http.StatusBadRequest, "invalid request: user does not exist", "", nil}
+	errHTTPBadRequestTwilioDisabled                  = &errHTTP{40032, http.StatusBadRequest, "invalid request: Calling is disabled", "https://ntfy.sh/docs/publish/#phone-calls", nil}
+	errHTTPBadRequestPhoneNumberInvalid              = &errHTTP{40033, http.StatusBadRequest, "invalid request: phone number invalid", "https://ntfy.sh/docs/publish/#phone-calls", nil}
 	errHTTPNotFound                                  = &errHTTP{40401, http.StatusNotFound, "page not found", "", nil}
 	errHTTPUnauthorized                              = &errHTTP{40101, http.StatusUnauthorized, "unauthorized", "https://ntfy.sh/docs/publish/#authentication", nil}
 	errHTTPForbidden                                 = &errHTTP{40301, http.StatusForbidden, "forbidden", "https://ntfy.sh/docs/publish/#authentication", nil}

server/server.go

@@ -82,6 +82,8 @@ var (
 	apiHealthPath                                        = "/v1/health"
 	apiStatsPath                                         = "/v1/stats"
 	apiTiersPath                                         = "/v1/tiers"
+	apiUsersPath                                         = "/v1/users"
+	apiUsersAccessPath                                   = "/v1/users/access"
 	apiAccountPath                                       = "/v1/account"
 	apiAccountTokenPath                                  = "/v1/account/token"
 	apiAccountPasswordPath                               = "/v1/account/password"
@@ -413,6 +415,16 @@ func (s *Server) handleInternal(w http.ResponseWriter, r *http.Request, v *visit
 		return s.handleHealth(w, r, v)
 	} else if r.Method == http.MethodGet && r.URL.Path == webConfigPath {
 		return s.ensureWebEnabled(s.handleWebConfig)(w, r, v)
+	} else if r.Method == http.MethodGet && r.URL.Path == apiUsersPath {
+		return s.ensureAdmin(s.handleUsersGet)(w, r, v)
+	} else if r.Method == http.MethodPut && r.URL.Path == apiUsersPath {
+		return s.ensureAdmin(s.handleUsersAdd)(w, r, v)
+	} else if r.Method == http.MethodDelete && r.URL.Path == apiUsersPath {
+		return s.ensureAdmin(s.handleUsersDelete)(w, r, v)
+	} else if (r.Method == http.MethodPut || r.Method == http.MethodPost) && r.URL.Path == apiUsersAccessPath {
+		return s.ensureAdmin(s.handleAccessAllow)(w, r, v)
+	} else if r.Method == http.MethodDelete && r.URL.Path == apiUsersAccessPath {
+		return s.ensureAdmin(s.handleAccessReset)(w, r, v)
 	} else if r.Method == http.MethodPost && r.URL.Path == apiAccountPath {
 		return s.ensureUserManager(s.handleAccountCreate)(w, r, v)
 	} else if r.Method == http.MethodGet && r.URL.Path == apiAccountPath {
@@ -651,6 +663,9 @@ func (s *Server) handleFile(w http.ResponseWriter, r *http.Request, v *visitor)
 		return err
 	}
 	defer f.Close()
+	if m.Attachment.Name != "" {
+		w.Header().Set("Content-Disposition", "attachment; filename="+strconv.Quote(m.Attachment.Name))
+	}
 	_, err = io.Copy(util.NewContentTypeWriter(w, r.URL.Path), f)
 	return err
 }
@@ -1221,7 +1236,7 @@ func (s *Server) handleSubscribeWS(w http.ResponseWriter, r *http.Request, v *vi
 	}
 	defer conn.Close()
 
-	// Subscription connections can be canceled externally, see topic.CancelSubscribers
+	// Subscription connections can be canceled externally, see topic.CancelSubscribersExceptUser
 	cancelCtx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
@@ -1463,6 +1478,7 @@ func (s *Server) handleOptions(w http.ResponseWriter, _ *http.Request, _ *visito
 	return nil
 }
 
+// topicFromPath returns the topic from a root path (e.g. /mytopic), creating it if it doesn't exist.
 func (s *Server) topicFromPath(path string) (*topic, error) {
 	parts := strings.Split(path, "/")
 	if len(parts) < 2 {
@@ -1471,6 +1487,7 @@ func (s *Server) topicFromPath(path string) (*topic, error) {
 	return s.topicFromID(parts[1])
 }
 
+// topicsFromPath returns the topic from a root path (e.g. /mytopic,mytopic2), creating it if it doesn't exist.
 func (s *Server) topicsFromPath(path string) ([]*topic, string, error) {
 	parts := strings.Split(path, "/")
 	if len(parts) < 2 {
@@ -1484,6 +1501,7 @@ func (s *Server) topicsFromPath(path string) ([]*topic, string, error) {
 	return topics, parts[1], nil
 }
 
+// topicsFromIDs returns the topics with the given IDs, creating them if they don't exist.
 func (s *Server) topicsFromIDs(ids ...string) ([]*topic, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
@@ -1503,6 +1521,7 @@ func (s *Server) topicsFromIDs(ids ...string) ([]*topic, error) {
 	return topics, nil
 }
 
+// topicFromID returns the topic with the given ID, creating it if it doesn't exist.
 func (s *Server) topicFromID(id string) (*topic, error) {
 	topics, err := s.topicsFromIDs(id)
 	if err != nil {
@@ -1511,6 +1530,23 @@ func (s *Server) topicFromID(id string) (*topic, error) {
 	return topics[0], nil
 }
 
+// topicsFromPattern returns a list of topics matching the given pattern, but it does not create them.
+func (s *Server) topicsFromPattern(pattern string) ([]*topic, error) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	patternRegexp, err := regexp.Compile("^" + strings.ReplaceAll(pattern, "*", ".*") + "$")
+	if err != nil {
+		return nil, err
+	}
+	topics := make([]*topic, 0)
+	for _, t := range s.topics {
+		if patternRegexp.MatchString(t.ID) {
+			topics = append(topics, t)
+		}
+	}
+	return topics, nil
+}
+
 func (s *Server) runSMTPServer() error {
 	s.smtpServerBackend = newMailBackend(s.config, s.handle)
 	s.smtpServer = smtp.NewServer(s.smtpServerBackend)

server/server_account.go

@@ -454,7 +454,7 @@ func (s *Server) handleAccountReservationAdd(w http.ResponseWriter, r *http.Requ
 	if err != nil {
 		return err
 	}
-	t.CancelSubscribers(u.ID)
+	t.CancelSubscribersExceptUser(u.ID)
 	return s.writeJSON(w, newSuccessResponse())
 }
 

server/server_admin.go

@@ -0,0 +1,143 @@
+package server
+
+import (
+	"heckel.io/ntfy/user"
+	"net/http"
+)
+
+func (s *Server) handleUsersGet(w http.ResponseWriter, r *http.Request, v *visitor) error {
+	users, err := s.userManager.Users()
+	if err != nil {
+		return err
+	}
+	grants, err := s.userManager.AllGrants()
+	if err != nil {
+		return err
+	}
+	usersResponse := make([]*apiUserResponse, len(users))
+	for i, u := range users {
+		tier := ""
+		if u.Tier != nil {
+			tier = u.Tier.Code
+		}
+		userGrants := make([]*apiUserGrantResponse, len(grants[u.ID]))
+		for i, g := range grants[u.ID] {
+			userGrants[i] = &apiUserGrantResponse{
+				Topic:      g.TopicPattern,
+				Permission: g.Allow.String(),
+			}
+		}
+		usersResponse[i] = &apiUserResponse{
+			Username: u.Name,
+			Role:     string(u.Role),
+			Tier:     tier,
+			Grants:   userGrants,
+		}
+	}
+	return s.writeJSON(w, usersResponse)
+}
+
+func (s *Server) handleUsersAdd(w http.ResponseWriter, r *http.Request, v *visitor) error {
+	req, err := readJSONWithLimit[apiUserAddRequest](r.Body, jsonBodyBytesLimit, false)
+	if err != nil {
+		return err
+	} else if !user.AllowedUsername(req.Username) || req.Password == "" {
+		return errHTTPBadRequest.Wrap("username invalid, or password missing")
+	}
+	u, err := s.userManager.User(req.Username)
+	if err != nil && err != user.ErrUserNotFound {
+		return err
+	} else if u != nil {
+		return errHTTPConflictUserExists
+	}
+	var tier *user.Tier
+	if req.Tier != "" {
+		tier, err = s.userManager.Tier(req.Tier)
+		if err == user.ErrTierNotFound {
+			return errHTTPBadRequestTierInvalid
+		} else if err != nil {
+			return err
+		}
+	}
+	if err := s.userManager.AddUser(req.Username, req.Password, user.RoleUser); err != nil {
+		return err
+	}
+	if tier != nil {
+		if err := s.userManager.ChangeTier(req.Username, req.Tier); err != nil {
+			return err
+		}
+	}
+	return s.writeJSON(w, newSuccessResponse())
+}
+
+func (s *Server) handleUsersDelete(w http.ResponseWriter, r *http.Request, v *visitor) error {
+	req, err := readJSONWithLimit[apiUserDeleteRequest](r.Body, jsonBodyBytesLimit, false)
+	if err != nil {
+		return err
+	}
+	u, err := s.userManager.User(req.Username)
+	if err == user.ErrUserNotFound {
+		return errHTTPBadRequestUserNotFound
+	} else if err != nil {
+		return err
+	} else if !u.IsUser() {
+		return errHTTPUnauthorized.Wrap("can only remove regular users from API")
+	}
+	if err := s.userManager.RemoveUser(req.Username); err != nil {
+		return err
+	}
+	if err := s.killUserSubscriber(u, "*"); err != nil { // FIXME super inefficient
+		return err
+	}
+	return s.writeJSON(w, newSuccessResponse())
+}
+
+func (s *Server) handleAccessAllow(w http.ResponseWriter, r *http.Request, v *visitor) error {
+	req, err := readJSONWithLimit[apiAccessAllowRequest](r.Body, jsonBodyBytesLimit, false)
+	if err != nil {
+		return err
+	}
+	_, err = s.userManager.User(req.Username)
+	if err == user.ErrUserNotFound {
+		return errHTTPBadRequestUserNotFound
+	} else if err != nil {
+		return err
+	}
+	permission, err := user.ParsePermission(req.Permission)
+	if err != nil {
+		return errHTTPBadRequestPermissionInvalid
+	}
+	if err := s.userManager.AllowAccess(req.Username, req.Topic, permission); err != nil {
+		return err
+	}
+	return s.writeJSON(w, newSuccessResponse())
+}
+
+func (s *Server) handleAccessReset(w http.ResponseWriter, r *http.Request, v *visitor) error {
+	req, err := readJSONWithLimit[apiAccessResetRequest](r.Body, jsonBodyBytesLimit, false)
+	if err != nil {
+		return err
+	}
+	u, err := s.userManager.User(req.Username)
+	if err != nil {
+		return err
+	}
+	if err := s.userManager.ResetAccess(req.Username, req.Topic); err != nil {
+		return err
+	}
+	if err := s.killUserSubscriber(u, req.Topic); err != nil { // This may be a pattern
+		return err
+	}
+	return s.writeJSON(w, newSuccessResponse())
+}
+
+func (s *Server) killUserSubscriber(u *user.User, topicPattern string) error {
+	topics, err := s.topicsFromPattern(topicPattern)
+	if err != nil {
+		return err
+	}
+	for _, t := range topics {
+		t.CancelSubscriberUser(u.ID)
+	}
+	return nil
+}

server/server_admin_test.go

@@ -0,0 +1,181 @@
+package server
+
+import (
+	"github.com/stretchr/testify/require"
+	"heckel.io/ntfy/user"
+	"heckel.io/ntfy/util"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestUser_AddRemove(t *testing.T) {
+	s := newTestServer(t, newTestConfigWithAuthFile(t))
+	defer s.closeDatabases()
+
+	// Create admin, tier
+	require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleAdmin))
+	require.Nil(t, s.userManager.AddTier(&user.Tier{
+		Code: "tier1",
+	}))
+
+	// Create user via API
+	rr := request(t, s, "PUT", "/v1/users", `{"username": "ben", "password":"ben"}`, map[string]string{
+		"Authorization": util.BasicAuth("phil", "phil"),
+	})
+	require.Equal(t, 200, rr.Code)
+
+	// Create user with tier via API
+	rr = request(t, s, "PUT", "/v1/users", `{"username": "emma", "password":"emma", "tier": "tier1"}`, map[string]string{
+		"Authorization": util.BasicAuth("phil", "phil"),
+	})
+	require.Equal(t, 200, rr.Code)
+
+	// Check users
+	users, err := s.userManager.Users()
+	require.Nil(t, err)
+	require.Equal(t, 4, len(users))
+	require.Equal(t, "phil", users[0].Name)
+	require.Equal(t, "ben", users[1].Name)
+	require.Equal(t, user.RoleUser, users[1].Role)
+	require.Nil(t, users[1].Tier)
+	require.Equal(t, "emma", users[2].Name)
+	require.Equal(t, user.RoleUser, users[2].Role)
+	require.Equal(t, "tier1", users[2].Tier.Code)
+	require.Equal(t, user.Everyone, users[3].Name)
+
+	// Delete user via API
+	rr = request(t, s, "DELETE", "/v1/users", `{"username": "ben"}`, map[string]string{
+		"Authorization": util.BasicAuth("phil", "phil"),
+	})
+	require.Equal(t, 200, rr.Code)
+}
+
+func TestUser_AddRemove_Failures(t *testing.T) {
+	s := newTestServer(t, newTestConfigWithAuthFile(t))
+	defer s.closeDatabases()
+
+	// Create admin
+	require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleAdmin))
+	require.Nil(t, s.userManager.AddUser("ben", "ben", user.RoleUser))
+
+	// Cannot create user with invalid username
+	rr := request(t, s, "PUT", "/v1/users", `{"username": "not valid", "password":"ben"}`, map[string]string{
+		"Authorization": util.BasicAuth("phil", "phil"),
+	})
+	require.Equal(t, 400, rr.Code)
+
+	// Cannot create user if user already exists
+	rr = request(t, s, "PUT", "/v1/users", `{"username": "phil", "password":"phil"}`, map[string]string{
+		"Authorization": util.BasicAuth("phil", "phil"),
+	})
+	require.Equal(t, 40901, toHTTPError(t, rr.Body.String()).Code)
+
+	// Cannot create user with invalid tier
+	rr = request(t, s, "PUT", "/v1/users", `{"username": "emma", "password":"emma", "tier": "invalid"}`, map[string]string{
+		"Authorization": util.BasicAuth("phil", "phil"),
+	})
+	require.Equal(t, 40030, toHTTPError(t, rr.Body.String()).Code)
+
+	// Cannot delete user as non-admin
+	rr = request(t, s, "DELETE", "/v1/users", `{"username": "ben"}`, map[string]string{
+		"Authorization": util.BasicAuth("ben", "ben"),
+	})
+	require.Equal(t, 401, rr.Code)
+
+	// Delete user via API
+	rr = request(t, s, "DELETE", "/v1/users", `{"username": "ben"}`, map[string]string{
+		"Authorization": util.BasicAuth("phil", "phil"),
+	})
+	require.Equal(t, 200, rr.Code)
+}
+
+func TestAccess_AllowReset(t *testing.T) {
+	c := newTestConfigWithAuthFile(t)
+	c.AuthDefault = user.PermissionDenyAll
+	s := newTestServer(t, c)
+	defer s.closeDatabases()
+
+	// User and admin
+	require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleAdmin))
+	require.Nil(t, s.userManager.AddUser("ben", "ben", user.RoleUser))
+
+	// Subscribing not allowed
+	rr := request(t, s, "GET", "/gold/json?poll=1", "", map[string]string{
+		"Authorization": util.BasicAuth("ben", "ben"),
+	})
+	require.Equal(t, 403, rr.Code)
+
+	// Grant access
+	rr = request(t, s, "POST", "/v1/users/access", `{"username": "ben", "topic":"gold", "permission":"ro"}`, map[string]string{
+		"Authorization": util.BasicAuth("phil", "phil"),
+	})
+	require.Equal(t, 200, rr.Code)
+
+	// Now subscribing is allowed
+	rr = request(t, s, "GET", "/gold/json?poll=1", "", map[string]string{
+		"Authorization": util.BasicAuth("ben", "ben"),
+	})
+	require.Equal(t, 200, rr.Code)
+
+	// Reset access
+	rr = request(t, s, "DELETE", "/v1/users/access", `{"username": "ben", "topic":"gold"}`, map[string]string{
+		"Authorization": util.BasicAuth("phil", "phil"),
+	})
+	require.Equal(t, 200, rr.Code)
+
+	// Subscribing not allowed (again)
+	rr = request(t, s, "GET", "/gold/json?poll=1", "", map[string]string{
+		"Authorization": util.BasicAuth("ben", "ben"),
+	})
+	require.Equal(t, 403, rr.Code)
+}
+
+func TestAccess_AllowReset_NonAdminAttempt(t *testing.T) {
+	c := newTestConfigWithAuthFile(t)
+	c.AuthDefault = user.PermissionDenyAll
+	s := newTestServer(t, c)
+	defer s.closeDatabases()
+
+	// User
+	require.Nil(t, s.userManager.AddUser("ben", "ben", user.RoleUser))
+
+	// Grant access fails, because non-admin
+	rr := request(t, s, "POST", "/v1/users/access", `{"username": "ben", "topic":"gold", "permission":"ro"}`, map[string]string{
+		"Authorization": util.BasicAuth("ben", "ben"),
+	})
+	require.Equal(t, 401, rr.Code)
+}
+
+func TestAccess_AllowReset_KillConnection(t *testing.T) {
+	c := newTestConfigWithAuthFile(t)
+	c.AuthDefault = user.PermissionDenyAll
+	s := newTestServer(t, c)
+	defer s.closeDatabases()
+
+	// User and admin, grant access to "gol*" topics
+	require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleAdmin))
+	require.Nil(t, s.userManager.AddUser("ben", "ben", user.RoleUser))
+	require.Nil(t, s.userManager.AllowAccess("ben", "gol*", user.PermissionRead)) // Wildcard!
+
+	start, timeTaken := time.Now(), atomic.Int64{}
+	go func() {
+		rr := request(t, s, "GET", "/gold/json", "", map[string]string{
+			"Authorization": util.BasicAuth("ben", "ben"),
+		})
+		require.Equal(t, 200, rr.Code)
+		timeTaken.Store(time.Since(start).Milliseconds())
+	}()
+	time.Sleep(500 * time.Millisecond)
+
+	// Reset access
+	rr := request(t, s, "DELETE", "/v1/users/access", `{"username": "ben", "topic":"gol*"}`, map[string]string{
+		"Authorization": util.BasicAuth("phil", "phil"),
+	})
+	require.Equal(t, 200, rr.Code)
+
+	// Wait for connection to be killed; this will fail if the connection is never killed
+	waitFor(t, func() bool {
+		return timeTaken.Load() >= 500
+	})
+}

server/server_middleware.go

@@ -76,6 +76,15 @@ func (s *Server) ensureUser(next handleFunc) handleFunc {
 	})
 }
 
+func (s *Server) ensureAdmin(next handleFunc) handleFunc {
+	return s.ensureUserManager(func(w http.ResponseWriter, r *http.Request, v *visitor) error {
+		if !v.User().IsAdmin() {
+			return errHTTPUnauthorized
+		}
+		return next(w, r, v)
+	})
+}
+
 func (s *Server) ensurePaymentsEnabled(next handleFunc) handleFunc {
 	return func(w http.ResponseWriter, r *http.Request, v *visitor) error {
 		if s.config.StripeSecretKey == "" || s.stripe == nil {

server/smtp_sender.go

@@ -4,14 +4,15 @@ import (
 	_ "embed" // required by go:embed
 	"encoding/json"
 	"fmt"
-	"heckel.io/ntfy/log"
-	"heckel.io/ntfy/util"
 	"mime"
 	"net"
 	"net/smtp"
 	"strings"
 	"sync"
 	"time"
+
+	"heckel.io/ntfy/log"
+	"heckel.io/ntfy/util"
 )
 
 type mailer interface {
@@ -131,31 +132,23 @@ This message was sent by {ip} at {time} via {topicURL}`
 }
 
 var (
-	//go:embed "mailer_emoji.json"
+	//go:embed "mailer_emoji_map.json"
 	emojisJSON string
 )
 
-type emoji struct {
-	Emoji   string   `json:"emoji"`
-	Aliases []string `json:"aliases"`
-}
-
 func toEmojis(tags []string) (emojisOut []string, tagsOut []string, err error) {
-	var emojis []emoji
-	if err = json.Unmarshal([]byte(emojisJSON), &emojis); err != nil {
+	var emojiMap map[string]string
+	if err = json.Unmarshal([]byte(emojisJSON), &emojiMap); err != nil {
 		return nil, nil, err
 	}
 	tagsOut = make([]string, 0)
 	emojisOut = make([]string, 0)
-nextTag:
-	for _, t := range tags { // TODO Super inefficient; we should just create a .json file with a map
-		for _, e := range emojis {
-			if util.Contains(e.Aliases, t) {
-				emojisOut = append(emojisOut, e.Emoji)
-				continue nextTag
-			}
+	for _, t := range tags {
+		if emoji, ok := emojiMap[t]; ok {
+			emojisOut = append(emojisOut, emoji)
+		} else {
+			tagsOut = append(tagsOut, t)
 		}
-		tagsOut = append(tagsOut, t)
 	}
 	return
 }

server/topic.go

@@ -141,24 +141,40 @@ func (t *topic) Keepalive() {
 	t.lastAccess = time.Now()
 }
 
-// CancelSubscribers calls the cancel function for all subscribers, forcing
-func (t *topic) CancelSubscribers(exceptUserID string) {
+// CancelSubscribersExceptUser calls the cancel function for all subscribers, forcing
+func (t *topic) CancelSubscribersExceptUser(exceptUserID string) {
 	t.mu.Lock()
 	defer t.mu.Unlock()
 	for _, s := range t.subscribers {
 		if s.userID != exceptUserID {
-			log.
-				Tag(tagSubscribe).
-				With(t).
-				Fields(log.Context{
-					"user_id": s.userID,
-				}).
-				Debug("Canceling subscriber %s", s.userID)
-			s.cancel()
+			t.cancelUserSubscriber(s)
 		}
 	}
 }
 
+// CancelSubscriberUser kills the subscriber with the given user ID
+func (t *topic) CancelSubscriberUser(userID string) {
+	t.mu.RLock()
+	defer t.mu.RUnlock()
+	for _, s := range t.subscribers {
+		if s.userID == userID {
+			t.cancelUserSubscriber(s)
+			return
+		}
+	}
+}
+
+func (t *topic) cancelUserSubscriber(s *topicSubscriber) {
+	log.
+		Tag(tagSubscribe).
+		With(t).
+		Fields(log.Context{
+			"user_id": s.userID,
+		}).
+		Debug("Canceling subscriber with user ID %s", s.userID)
+	s.cancel()
+}
+
 func (t *topic) Context() log.Context {
 	t.mu.RLock()
 	defer t.mu.RUnlock()

server/topic_test.go

@@ -9,7 +9,7 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestTopic_CancelSubscribers(t *testing.T) {
+func TestTopic_CancelSubscribersExceptUser(t *testing.T) {
 	t.Parallel()
 
 	subFn := func(v *visitor, msg *message) error {
@@ -27,11 +27,34 @@ func TestTopic_CancelSubscribers(t *testing.T) {
 	to.Subscribe(subFn, "", cancelFn1)
 	to.Subscribe(subFn, "u_phil", cancelFn2)
 
-	to.CancelSubscribers("u_phil")
+	to.CancelSubscribersExceptUser("u_phil")
 	require.True(t, canceled1.Load())
 	require.False(t, canceled2.Load())
 }
 
+func TestTopic_CancelSubscribersUser(t *testing.T) {
+	t.Parallel()
+
+	subFn := func(v *visitor, msg *message) error {
+		return nil
+	}
+	canceled1 := atomic.Bool{}
+	cancelFn1 := func() {
+		canceled1.Store(true)
+	}
+	canceled2 := atomic.Bool{}
+	cancelFn2 := func() {
+		canceled2.Store(true)
+	}
+	to := newTopic("mytopic")
+	to.Subscribe(subFn, "u_another", cancelFn1)
+	to.Subscribe(subFn, "u_phil", cancelFn2)
+
+	to.CancelSubscriberUser("u_phil")
+	require.False(t, canceled1.Load())
+	require.True(t, canceled2.Load())
+}
+
 func TestTopic_Keepalive(t *testing.T) {
 	t.Parallel()
 

server/types.go

@@ -244,6 +244,40 @@ type apiStatsResponse struct {
 	MessagesRate float64 `json:"messages_rate"` // Average number of messages per second
 }
 
+type apiUserAddRequest struct {
+	Username string `json:"username"`
+	Password string `json:"password"`
+	Tier     string `json:"tier"`
+	// Do not add 'role' here. We don't want to add admins via the API.
+}
+
+type apiUserResponse struct {
+	Username string                  `json:"username"`
+	Role     string                  `json:"role"`
+	Tier     string                  `json:"tier,omitempty"`
+	Grants   []*apiUserGrantResponse `json:"grants,omitempty"`
+}
+
+type apiUserGrantResponse struct {
+	Topic      string `json:"topic"` // This may be a pattern
+	Permission string `json:"permission"`
+}
+
+type apiUserDeleteRequest struct {
+	Username string `json:"username"`
+}
+
+type apiAccessAllowRequest struct {
+	Username   string `json:"username"`
+	Topic      string `json:"topic"` // This may be a pattern
+	Permission string `json:"permission"`
+}
+
+type apiAccessResetRequest struct {
+	Username string `json:"username"`
+	Topic    string `json:"topic"`
+}
+
 type apiAccountCreateRequest struct {
 	Username string `json:"username"`
 	Password string `json:"password"`