gogrlx/nats-server
1
0
Fork 0
mirror of https://github.com/gogrlx/nats-server.git synced 2026-04-02 03:38:42 -07:00
Code Issues Packages Projects Releases Wiki Activity

Allow dynamic publish permissions based on reply subjects of received msgs

Browse Source 
Signed-off-by: Derek Collison <derek@nats.io>
This commit is contained in:
Derek Collison
2019-07-24 17:52:18 -07:00
parent 9cf534bc5e
commit 495a1a7ec3
10 changed files with 518 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
server/auth.go
View File

@@ -1,4 +1,4 @@
// Copyright 2012-2018 The NATS Authors
// Copyright 2012-2019 The NATS Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
@@ -18,6 +18,7 @@ import (
"encoding/base64"
"net"
"strings"
"time"
"github.com/nats-io/jwt"
"github.com/nats-io/nkeys"
@@ -89,11 +90,19 @@ type SubjectPermission struct {
Deny []string `json:"deny,omitempty"`
}
// ResponsePermission can be used to allow responses to any reply subject
// that is received on a valid subscription.
type ResponsePermission struct {
MaxMsgs int `json:"max"`
Expires time.Duration `json:"ttl"`
}
// Permissions are the allowed subjects on a per
// publish or subscribe basis.
type Permissions struct {
Publish *SubjectPermission `json:"publish"`
Subscribe *SubjectPermission `json:"subscribe"`
Publish *SubjectPermission `json:"publish"`
Subscribe *SubjectPermission `json:"subscribe"`
Response *ResponsePermission `json:"responses,omitempty"`
}
// RoutePermissions are similar to user permissions
@@ -134,6 +143,12 @@ func (p *Permissions) clone() *Permissions {
if p.Subscribe != nil {
clone.Subscribe = p.Subscribe.clone()
}
if p.Response != nil {
clone.Response = &ResponsePermission{
MaxMsgs: p.Response.MaxMsgs,
Expires: p.Response.Expires,
}
}
return clone
}

149
server/client.go
View File

@@ -157,33 +157,35 @@ const (
type client struct {
// Here first because of use of atomics, and memory alignment.
stats
mpay int32
msubs int32
mcl int32
mu sync.Mutex
kind int
cid uint64
opts clientOpts
start time.Time
nonce []byte
nc net.Conn
ncs string
out outbound
srv *Server
acc *Account
user *NkeyUser
host string
port uint16
subs map[string]*subscription
perms *permissions
mperms *msgDeny
darray []string
in readCache
pcd map[*client]struct{}
atmr *time.Timer
ping pinfo
msgb [msgScratchSize]byte
last time.Time
mpay int32
msubs int32
mcl int32
mu sync.Mutex
kind int
cid uint64
opts clientOpts
start time.Time
nonce []byte
nc net.Conn
ncs string
out outbound
srv *Server
acc *Account
user *NkeyUser
host string
port uint16
subs map[string]*subscription
perms *permissions
replies map[string]*resp
rcheck time.Time
mperms *msgDeny
darray []string
in readCache
pcd map[*client]struct{}
atmr *time.Timer
ping pinfo
msgb [msgScratchSize]byte
last time.Time
parseState
rtt time.Duration
@@ -230,12 +232,21 @@ type perm struct {
allow *Sublist
deny *Sublist
}
type permissions struct {
sub perm
pub perm
resp *ResponsePermission
pcache map[string]bool
}
// This is used to dynamically track responses and reply subjects
// for dynamic permissioning.
type resp struct {
t time.Time
n int
}
// msgDeny is used when a user permission for subscriptions has a deny
// clause but a subscription could be made that is of broader scope.
// e.g. deny = "foo", but user subscribes to "*". That subscription should
@@ -259,6 +270,8 @@ const (
maxPermCacheSize = 128
pruneSize = 32
routeTargetInit = 8
replyPermLimit = 4096
replyCheckMin = 30 * time.Second
)
// Used in readloop to cache hot subject lookups and group statistics.
@@ -536,10 +549,9 @@ func (c *client) RegisterUser(user *User) {
// Reset perms to nil in case client previously had them.
c.perms = nil
c.mperms = nil
c.mu.Unlock()
return
} else {
c.setPermissions(user.Permissions)
}
c.setPermissions(user.Permissions)
c.mu.Unlock()
}
@@ -580,7 +592,7 @@ func (c *client) setPermissions(perms *Permissions) {
// Loop over publish permissions
if perms.Publish != nil {
if len(perms.Publish.Allow) > 0 {
if perms.Publish.Allow != nil {
c.perms.pub.allow = NewSublistWithCache()
}
for _, pubSubject := range perms.Publish.Allow {
@@ -596,6 +608,14 @@ func (c *client) setPermissions(perms *Permissions) {
}
}
// Check if we are allowed to send responses.
if perms.Response != nil {
rp := *perms.Response
c.perms.resp = &rp
c.replies = make(map[string]*resp)
c.rcheck = time.Now()
}
// Loop over subscribe permissions
if perms.Subscribe != nil {
if len(perms.Subscribe.Allow) > 0 {
@@ -2131,6 +2151,15 @@ func (c *client) deliverMsg(sub *subscription, mh, msg []byte) bool {
client.out.pm++
// If we are tracking dynamic publish permissions that track reply subjects,
// do that accounting here. We only look at client.replies which will be non-nil.
if client.replies != nil && len(c.pa.reply) > 0 {
client.replies[string(c.pa.reply)] = &resp{time.Now(), 0}
if len(client.replies) > replyPermLimit {
client.pruneReplyPerms()
}
}
// Check outbound threshold and queue IO flush if needed.
// This is specifically looking at situations where we are getting behind and may want
// to intervene before this producer goes back to top of readloop. We are in the producer's
@@ -2156,6 +2185,27 @@ func (c *client) deliverMsg(sub *subscription, mh, msg []byte) bool {
return true
}
// pruneReplyPerms will remove any stale or expired entries
// in our reply cache. We make sure to not check too often.
func (c *client) pruneReplyPerms() {
// Make sure we do not check too often.
if c.perms.resp == nil || time.Since(c.rcheck) < replyCheckMin {
return
}
mm := c.perms.resp.MaxMsgs
ttl := c.perms.resp.Expires
c.rcheck = time.Now()
for k, resp := range c.replies {
if mm > 0 && resp.n >= mm {
delete(c.replies, k)
} else if ttl > 0 && c.rcheck.Sub(resp.t) > ttl {
delete(c.replies, k)
}
}
}
// pruneDenyCache will prune the deny cache via randomly
// deleting items. Doing so pruneSize items at a time.
// Lock must be held for this one since it is shared under
@@ -2183,7 +2233,14 @@ func (c *client) prunePubPermsCache() {
}
// pubAllowed checks on publish permissioning.
// Lock should not be held.
func (c *client) pubAllowed(subject string) bool {
return c.pubAllowedFullCheck(subject, true)
}
// pubAllowedFullCheck checks on all publish permissioning depending
// on the flag for dynamic reply permissions.
func (c *client) pubAllowedFullCheck(subject string, fullCheck bool) bool {
if c.perms == nil || (c.perms.pub.allow == nil && c.perms.pub.deny == nil) {
return true
}
@@ -2205,11 +2262,31 @@ func (c *client) pubAllowed(subject string) bool {
r := c.perms.pub.deny.Match(subject)
allowed = len(r.psubs) == 0
}
// Update our cache here.
c.perms.pcache[string(subject)] = allowed
// Prune if needed.
if len(c.perms.pcache) > maxPermCacheSize {
c.prunePubPermsCache()
// If we are currently not allowed but we are tracking reply subjects
// dynamically, check to see if we are allowed here Avoid pcache.
// We need to acquire the lock though.
if !allowed && fullCheck && c.perms.resp != nil {
c.mu.Lock()
if resp := c.replies[subject]; resp != nil {
resp.n++
// Check if we have sent too many responses.
if c.perms.resp.MaxMsgs > 0 && resp.n > c.perms.resp.MaxMsgs {
delete(c.replies, subject)
} else if c.perms.resp.Expires > 0 && time.Since(resp.t) > c.perms.resp.Expires {
delete(c.replies, subject)
} else {
allowed = true
}
}
c.mu.Unlock()
} else {
// Update our cache here.
c.perms.pcache[string(subject)] = allowed
// Prune if needed.
if len(c.perms.pcache) > maxPermCacheSize {
c.prunePubPermsCache()
}
}
return allowed
}

15
server/configs/authorization.conf
View File

@@ -23,6 +23,19 @@ authorization {
subscribe = "PUBLIC.>"
}
# Service can listen on the request subject and respond to any
# received reply subject.
my_service = {
subscribe = "my.service.req"
publish_allow_responses: true
}
# Can support a map with max messages and expiration of the permission.
my_stream_service = {
subscribe = "my.service.req"
allow_responses: {max: 10, expires: "1m"}
}
# Default permissions if none presented. e.g. susan below.
default_permissions: $default_user
@@ -31,5 +44,7 @@ authorization {
{user: alice, password: foo, permissions: $super_user}
{user: bob, password: bar, permissions: $req_pub_user}
{user: susan, password: baz}
{user: svca, password: pc, permissions: $my_service}
{user: svcb, password: sam, permissions: $my_stream_service}
]
}

8
server/const.go
View File

@@ -163,4 +163,12 @@ const (
// DEFAULT_RTT_MEASUREMENT_INTERVAL is how often we want to measure RTT from
// this server to clients, routes, gateways or leafnode connections.
DEFAULT_RTT_MEASUREMENT_INTERVAL = time.Hour
// DEFAULT_ALLOW_RESPONSE_MAX_MSGS is the default number of responses allowed
// for a reply subject.
DEFAULT_ALLOW_RESPONSE_MAX_MSGS = 1
// DEFAULT_ALLOW_RESPONSE_EXPIRATION is the default time allowed for a given
// dynamic response permission.
DEFAULT_ALLOW_RESPONSE_EXPIRATION = 2 * time.Minute
)

81
server/opts.go
View File

@@ -859,6 +859,12 @@ func parseCluster(v interface{}, opts *Options, errors *[]error, warnings *[]err
*errors = append(*errors, err)
continue
}
// Dynamic response permissions do not make sense here.
if perms.Response != nil {
err := &configErr{tk, fmt.Sprintf("Cluster permissions do not support dynamic responses")}
*errors = append(*errors, err)
continue
}
// This will possibly override permissions that were define in auth block
setClusterPermissions(&opts.Cluster, perms)
default:
@@ -1909,26 +1915,49 @@ func parseUserPermissions(mv interface{}, errors, warnings *[]error) (*Permissio
return nil, &configErr{tk, fmt.Sprintf("Expected permissions to be a map/struct, got %+v", mv)}
}
for k, v := range pm {
tk, v = unwrapValue(v)
tk, mv = unwrapValue(v)
switch strings.ToLower(k) {
// For routes:
// Import is Publish
// Export is Subscribe
case "pub", "publish", "import":
perms, err := parseVariablePermissions(v, errors, warnings)
perms, err := parseVariablePermissions(mv, errors, warnings)
if err != nil {
*errors = append(*errors, err)
continue
}
p.Publish = perms
case "sub", "subscribe", "export":
perms, err := parseVariablePermissions(v, errors, warnings)
perms, err := parseVariablePermissions(mv, errors, warnings)
if err != nil {
*errors = append(*errors, err)
continue
}
p.Subscribe = perms
case "publish_allow_responses", "allow_responses":
rp := &ResponsePermission{
MaxMsgs: DEFAULT_ALLOW_RESPONSE_MAX_MSGS,
Expires: DEFAULT_ALLOW_RESPONSE_EXPIRATION,
}
// Try boolean first
responses, ok := mv.(bool)
if ok {
if responses {
p.Response = rp
}
} else {
p.Response = parseAllowResponses(v, errors, warnings)
}
if p.Response != nil {
if p.Publish == nil {
p.Publish = &SubjectPermission{}
}
if p.Publish.Allow == nil {
// We turn off the blanket allow statement.
p.Publish.Allow = []string{}
}
}
default:
if !tk.IsUsedVariable() {
err := &configErr{tk, fmt.Sprintf("Unknown field %q parsing permissions", k)}
@@ -1980,6 +2009,52 @@ func parseSubjects(v interface{}, errors, warnings *[]error) ([]string, error) {
return subjects, nil
}
// Helper function to parse a ResponsePermission.
func parseAllowResponses(v interface{}, errors, warnings *[]error) *ResponsePermission {
tk, v := unwrapValue(v)
// Check if this is a map.
pm, ok := v.(map[string]interface{})
if !ok {
err := &configErr{tk, "error parsing response permissions, expected a boolean or a map"}
*errors = append(*errors, err)
return nil
}
rp := &ResponsePermission{
MaxMsgs: DEFAULT_ALLOW_RESPONSE_MAX_MSGS,
Expires: DEFAULT_ALLOW_RESPONSE_EXPIRATION,
}
for k, v := range pm {
tk, v = unwrapValue(v)
switch strings.ToLower(k) {
case "max", "max_msgs", "max_messages", "max_responses":
rp.MaxMsgs = int(v.(int64))
case "expires", "expiration", "ttl":
wd, ok := v.(string)
if ok {
ttl, err := time.ParseDuration(wd)
if err != nil {
err := &configErr{tk, fmt.Sprintf("error parsing expires: %v", err)}
*errors = append(*errors, err)
return nil
}
rp.Expires = ttl
} else {
err := &configErr{tk, "error parsing expires, not a duration string"}
*errors = append(*errors, err)
return nil
}
default:
if !tk.IsUsedVariable() {
err := &configErr{tk, fmt.Sprintf("Unknown field %q parsing permissions", k)}
*errors = append(*errors, err)
}
}
}
return rp
}
// Helper function to parse old style authorization configs.
func parseOldPermissionStyle(v interface{}, errors, warnings *[]error) (*SubjectPermission, error) {
subjects, err := parseSubjects(v, errors, warnings)

262
server/opts_test.go
View File

@@ -76,7 +76,7 @@ func TestOptions_RandomPort(t *testing.T) {
if opts.Port != 0 {
t.Fatalf("Process of options should have resolved random port to "+
"zero.\nexpected: %d\ngot: %d\n", 0, opts.Port)
"zero.\nexpected: %d\ngot: %d", 0, opts.Port)
}
}
@@ -111,7 +111,7 @@ func TestConfigFile(t *testing.T) {
opts, err := ProcessConfigFile("./configs/test.conf")
if err != nil {
t.Fatalf("Received an error reading config file: %v\n", err)
t.Fatalf("Received an error reading config file: %v", err)
}
checkOptionsEqual(t, golden, opts)
@@ -129,7 +129,7 @@ func TestTLSConfigFile(t *testing.T) {
}
opts, err := ProcessConfigFile("./configs/tls.conf")
if err != nil {
t.Fatalf("Received an error reading config file: %v\n", err)
t.Fatalf("Received an error reading config file: %v", err)
}
tlsConfig := opts.TLSConfig
if tlsConfig == nil {
@@ -156,13 +156,13 @@ func TestTLSConfigFile(t *testing.T) {
}
cert := tlsConfig.Certificates[0].Leaf
if err := cert.VerifyHostname("127.0.0.1"); err != nil {
t.Fatalf("Could not verify hostname in certificate: %v\n", err)
t.Fatalf("Could not verify hostname in certificate: %v", err)
}
// Now test adding cipher suites.
opts, err = ProcessConfigFile("./configs/tls_ciphers.conf")
if err != nil {
t.Fatalf("Received an error reading config file: %v\n", err)
t.Fatalf("Received an error reading config file: %v", err)
}
tlsConfig = opts.TLSConfig
if tlsConfig == nil {
@@ -208,7 +208,7 @@ func TestTLSConfigFile(t *testing.T) {
// test on a file that will load the curve preference defaults
opts, err = ProcessConfigFile("./configs/tls_ciphers.conf")
if err != nil {
t.Fatalf("Received an error reading config file: %v\n", err)
t.Fatalf("Received an error reading config file: %v", err)
}
if !reflect.DeepEqual(opts.TLSConfig.CurvePreferences, defaultCurvePreferences()) {
@@ -269,7 +269,7 @@ func TestMergeOverrides(t *testing.T) {
}
fopts, err := ProcessConfigFile("./configs/test.conf")
if err != nil {
t.Fatalf("Received an error reading config file: %v\n", err)
t.Fatalf("Received an error reading config file: %v", err)
}
// Overrides via flags
@@ -345,7 +345,7 @@ func TestRouteFlagOverride(t *testing.T) {
fopts, err := ProcessConfigFile("./configs/srv_a.conf")
if err != nil {
t.Fatalf("Received an error reading config file: %v\n", err)
t.Fatalf("Received an error reading config file: %v", err)
}
// Overrides via flags
@@ -384,7 +384,7 @@ func TestClusterFlagsOverride(t *testing.T) {
fopts, err := ProcessConfigFile("./configs/srv_a.conf")
if err != nil {
t.Fatalf("Received an error reading config file: %v\n", err)
t.Fatalf("Received an error reading config file: %v", err)
}
// Overrides via flags
@@ -419,7 +419,7 @@ func TestRouteFlagOverrideWithMultiple(t *testing.T) {
fopts, err := ProcessConfigFile("./configs/srv_a.conf")
if err != nil {
t.Fatalf("Received an error reading config file: %v\n", err)
t.Fatalf("Received an error reading config file: %v", err)
}
// Overrides via flags
@@ -434,23 +434,23 @@ func TestRouteFlagOverrideWithMultiple(t *testing.T) {
func TestDynamicPortOnListen(t *testing.T) {
opts, err := ProcessConfigFile("./configs/listen-1.conf")
if err != nil {
t.Fatalf("Received an error reading config file: %v\n", err)
t.Fatalf("Received an error reading config file: %v", err)
}
if opts.Port != -1 {
t.Fatalf("Received incorrect port %v, expected -1\n", opts.Port)
t.Fatalf("Received incorrect port %v, expected -1", opts.Port)
}
if opts.HTTPPort != -1 {
t.Fatalf("Received incorrect monitoring port %v, expected -1\n", opts.HTTPPort)
t.Fatalf("Received incorrect monitoring port %v, expected -1", opts.HTTPPort)
}
if opts.HTTPSPort != -1 {
t.Fatalf("Received incorrect secure monitoring port %v, expected -1\n", opts.HTTPSPort)
t.Fatalf("Received incorrect secure monitoring port %v, expected -1", opts.HTTPSPort)
}
}
func TestListenConfig(t *testing.T) {
opts, err := ProcessConfigFile("./configs/listen.conf")
if err != nil {
t.Fatalf("Received an error reading config file: %v\n", err)
t.Fatalf("Received an error reading config file: %v", err)
}
setBaselineOptions(opts)
@@ -459,13 +459,13 @@ func TestListenConfig(t *testing.T) {
port := 4422
monHost := "127.0.0.1"
if opts.Host != host {
t.Fatalf("Received incorrect host %q, expected %q\n", opts.Host, host)
t.Fatalf("Received incorrect host %q, expected %q", opts.Host, host)
}
if opts.HTTPHost != monHost {
t.Fatalf("Received incorrect host %q, expected %q\n", opts.HTTPHost, monHost)
t.Fatalf("Received incorrect host %q, expected %q", opts.HTTPHost, monHost)
}
if opts.Port != port {
t.Fatalf("Received incorrect port %v, expected %v\n", opts.Port, port)
t.Fatalf("Received incorrect port %v, expected %v", opts.Port, port)
}
// Clustering
@@ -473,10 +473,10 @@ func TestListenConfig(t *testing.T) {
clusterPort := 4244
if opts.Cluster.Host != clusterHost {
t.Fatalf("Received incorrect cluster host %q, expected %q\n", opts.Cluster.Host, clusterHost)
t.Fatalf("Received incorrect cluster host %q, expected %q", opts.Cluster.Host, clusterHost)
}
if opts.Cluster.Port != clusterPort {
t.Fatalf("Received incorrect cluster port %v, expected %v\n", opts.Cluster.Port, clusterPort)
t.Fatalf("Received incorrect cluster port %v, expected %v", opts.Cluster.Port, clusterPort)
}
// HTTP
@@ -484,56 +484,56 @@ func TestListenConfig(t *testing.T) {
httpPort := 8422
if opts.HTTPHost != httpHost {
t.Fatalf("Received incorrect http host %q, expected %q\n", opts.HTTPHost, httpHost)
t.Fatalf("Received incorrect http host %q, expected %q", opts.HTTPHost, httpHost)
}
if opts.HTTPPort != httpPort {
t.Fatalf("Received incorrect http port %v, expected %v\n", opts.HTTPPort, httpPort)
t.Fatalf("Received incorrect http port %v, expected %v", opts.HTTPPort, httpPort)
}
// HTTPS
httpsPort := 9443
if opts.HTTPSPort != httpsPort {
t.Fatalf("Received incorrect https port %v, expected %v\n", opts.HTTPSPort, httpsPort)
t.Fatalf("Received incorrect https port %v, expected %v", opts.HTTPSPort, httpsPort)
}
}
func TestListenPortOnlyConfig(t *testing.T) {
opts, err := ProcessConfigFile("./configs/listen_port.conf")
if err != nil {
t.Fatalf("Received an error reading config file: %v\n", err)
t.Fatalf("Received an error reading config file: %v", err)
}
setBaselineOptions(opts)
port := 8922
if opts.Host != DEFAULT_HOST {
t.Fatalf("Received incorrect host %q, expected %q\n", opts.Host, DEFAULT_HOST)
t.Fatalf("Received incorrect host %q, expected %q", opts.Host, DEFAULT_HOST)
}
if opts.HTTPHost != DEFAULT_HOST {
t.Fatalf("Received incorrect host %q, expected %q\n", opts.Host, DEFAULT_HOST)
t.Fatalf("Received incorrect host %q, expected %q", opts.Host, DEFAULT_HOST)
}
if opts.Port != port {
t.Fatalf("Received incorrect port %v, expected %v\n", opts.Port, port)
t.Fatalf("Received incorrect port %v, expected %v", opts.Port, port)
}
}
func TestListenPortWithColonConfig(t *testing.T) {
opts, err := ProcessConfigFile("./configs/listen_port_with_colon.conf")
if err != nil {
t.Fatalf("Received an error reading config file: %v\n", err)
t.Fatalf("Received an error reading config file: %v", err)
}
setBaselineOptions(opts)
port := 8922
if opts.Host != DEFAULT_HOST {
t.Fatalf("Received incorrect host %q, expected %q\n", opts.Host, DEFAULT_HOST)
t.Fatalf("Received incorrect host %q, expected %q", opts.Host, DEFAULT_HOST)
}
if opts.HTTPHost != DEFAULT_HOST {
t.Fatalf("Received incorrect host %q, expected %q\n", opts.Host, DEFAULT_HOST)
t.Fatalf("Received incorrect host %q, expected %q", opts.Host, DEFAULT_HOST)
}
if opts.Port != port {
t.Fatalf("Received incorrect port %v, expected %v\n", opts.Port, port)
t.Fatalf("Received incorrect port %v, expected %v", opts.Port, port)
}
}
@@ -545,20 +545,20 @@ func TestListenMonitoringDefault(t *testing.T) {
host := "10.0.1.22"
if opts.Host != host {
t.Fatalf("Received incorrect host %q, expected %q\n", opts.Host, host)
t.Fatalf("Received incorrect host %q, expected %q", opts.Host, host)
}
if opts.HTTPHost != host {
t.Fatalf("Received incorrect host %q, expected %q\n", opts.Host, host)
t.Fatalf("Received incorrect host %q, expected %q", opts.Host, host)
}
if opts.Port != DEFAULT_PORT {
t.Fatalf("Received incorrect port %v, expected %v\n", opts.Port, DEFAULT_PORT)
t.Fatalf("Received incorrect port %v, expected %v", opts.Port, DEFAULT_PORT)
}
}
func TestMultipleUsersConfig(t *testing.T) {
opts, err := ProcessConfigFile("./configs/multiple_users.conf")
if err != nil {
t.Fatalf("Received an error reading config file: %v\n", err)
t.Fatalf("Received an error reading config file: %v", err)
}
setBaselineOptions(opts)
}
@@ -568,12 +568,12 @@ func TestMultipleUsersConfig(t *testing.T) {
func TestAuthorizationConfig(t *testing.T) {
opts, err := ProcessConfigFile("./configs/authorization.conf")
if err != nil {
t.Fatalf("Received an error reading config file: %v\n", err)
t.Fatalf("Received an error reading config file: %v", err)
}
setBaselineOptions(opts)
lu := len(opts.Users)
if lu != 3 {
t.Fatalf("Expected 3 users, got %d\n", lu)
if lu != 5 {
t.Fatalf("Expected 5 users, got %d", lu)
}
// Build a map
mu := make(map[string]*User)
@@ -584,69 +584,153 @@ func TestAuthorizationConfig(t *testing.T) {
// Alice
alice, ok := mu["alice"]
if !ok {
t.Fatalf("Expected to see user Alice\n")
t.Fatalf("Expected to see user Alice")
}
// Check for permissions details
if alice.Permissions == nil {
t.Fatalf("Expected Alice's permissions to be non-nil\n")
t.Fatalf("Expected Alice's permissions to be non-nil")
}
if alice.Permissions.Publish == nil {
t.Fatalf("Expected Alice's publish permissions to be non-nil\n")
t.Fatalf("Expected Alice's publish permissions to be non-nil")
}
if len(alice.Permissions.Publish.Allow) != 1 {
t.Fatalf("Expected Alice's publish permissions to have 1 element, got %d\n",
t.Fatalf("Expected Alice's publish permissions to have 1 element, got %d",
len(alice.Permissions.Publish.Allow))
}
pubPerm := alice.Permissions.Publish.Allow[0]
if pubPerm != "*" {
t.Fatalf("Expected Alice's publish permissions to be '*', got %q\n", pubPerm)
t.Fatalf("Expected Alice's publish permissions to be '*', got %q", pubPerm)
}
if alice.Permissions.Subscribe == nil {
t.Fatalf("Expected Alice's subscribe permissions to be non-nil\n")
t.Fatalf("Expected Alice's subscribe permissions to be non-nil")
}
if len(alice.Permissions.Subscribe.Allow) != 1 {
t.Fatalf("Expected Alice's subscribe permissions to have 1 element, got %d\n",
t.Fatalf("Expected Alice's subscribe permissions to have 1 element, got %d",
len(alice.Permissions.Subscribe.Allow))
}
subPerm := alice.Permissions.Subscribe.Allow[0]
if subPerm != ">" {
t.Fatalf("Expected Alice's subscribe permissions to be '>', got %q\n", subPerm)
t.Fatalf("Expected Alice's subscribe permissions to be '>', got %q", subPerm)
}
// Bob
bob, ok := mu["bob"]
if !ok {
t.Fatalf("Expected to see user Bob\n")
t.Fatalf("Expected to see user Bob")
}
if bob.Permissions == nil {
t.Fatalf("Expected Bob's permissions to be non-nil\n")
t.Fatalf("Expected Bob's permissions to be non-nil")
}
// Susan
susan, ok := mu["susan"]
if !ok {
t.Fatalf("Expected to see user Susan\n")
t.Fatalf("Expected to see user Susan")
}
if susan.Permissions == nil {
t.Fatalf("Expected Susan's permissions to be non-nil\n")
t.Fatalf("Expected Susan's permissions to be non-nil")
}
// Check susan closely since she inherited the default permissions.
if susan.Permissions == nil {
t.Fatalf("Expected Susan's permissions to be non-nil\n")
t.Fatalf("Expected Susan's permissions to be non-nil")
}
if susan.Permissions.Publish != nil {
t.Fatalf("Expected Susan's publish permissions to be nil\n")
t.Fatalf("Expected Susan's publish permissions to be nil")
}
if susan.Permissions.Subscribe == nil {
t.Fatalf("Expected Susan's subscribe permissions to be non-nil\n")
t.Fatalf("Expected Susan's subscribe permissions to be non-nil")
}
if len(susan.Permissions.Subscribe.Allow) != 1 {
t.Fatalf("Expected Susan's subscribe permissions to have 1 element, got %d\n",
t.Fatalf("Expected Susan's subscribe permissions to have 1 element, got %d",
len(susan.Permissions.Subscribe.Allow))
}
subPerm = susan.Permissions.Subscribe.Allow[0]
if subPerm != "PUBLIC.>" {
t.Fatalf("Expected Susan's subscribe permissions to be 'PUBLIC.>', got %q\n", subPerm)
t.Fatalf("Expected Susan's subscribe permissions to be 'PUBLIC.>', got %q", subPerm)
}
// Service A
svca, ok := mu["svca"]
if !ok {
t.Fatalf("Expected to see user Service A")
}
if svca.Permissions == nil {
t.Fatalf("Expected Service A's permissions to be non-nil")
}
if svca.Permissions.Subscribe == nil {
t.Fatalf("Expected Service A's subscribe permissions to be non-nil")
}
if len(svca.Permissions.Subscribe.Allow) != 1 {
t.Fatalf("Expected Service A's subscribe permissions to have 1 element, got %d",
len(svca.Permissions.Subscribe.Allow))
}
subPerm = svca.Permissions.Subscribe.Allow[0]
if subPerm != "my.service.req" {
t.Fatalf("Expected Service A's subscribe permissions to be 'my.service.req', got %q", subPerm)
}
// We want allow_responses to essentially set deny all, or allow none in this case.
if svca.Permissions.Publish == nil {
t.Fatalf("Expected Service A's publish permissions to be non-nil")
}
if len(svca.Permissions.Publish.Allow) != 0 {
t.Fatalf("Expected Service A's publish permissions to have no elements, got %d",
len(svca.Permissions.Publish.Allow))
}
// We should have a ResponsePermission present with default values.
if svca.Permissions.Response == nil {
t.Fatalf("Expected Service A's response permissions to be non-nil")
}
if svca.Permissions.Response.MaxMsgs != DEFAULT_ALLOW_RESPONSE_MAX_MSGS {
t.Fatalf("Expected Service A's response permissions of max msgs to be %d, got %d",
DEFAULT_ALLOW_RESPONSE_MAX_MSGS, svca.Permissions.Response.MaxMsgs,
)
}
if svca.Permissions.Response.Expires != DEFAULT_ALLOW_RESPONSE_EXPIRATION {
t.Fatalf("Expected Service A's response permissions of expiration to be %v, got %v",
DEFAULT_ALLOW_RESPONSE_EXPIRATION, svca.Permissions.Response.Expires,
)
}
// Service B
svcb, ok := mu["svcb"]
if !ok {
t.Fatalf("Expected to see user Service B")
}
if svcb.Permissions == nil {
t.Fatalf("Expected Service B's permissions to be non-nil")
}
if svcb.Permissions.Subscribe == nil {
t.Fatalf("Expected Service B's subscribe permissions to be non-nil")
}
if len(svcb.Permissions.Subscribe.Allow) != 1 {
t.Fatalf("Expected Service B's subscribe permissions to have 1 element, got %d",
len(svcb.Permissions.Subscribe.Allow))
}
subPerm = svcb.Permissions.Subscribe.Allow[0]
if subPerm != "my.service.req" {
t.Fatalf("Expected Service B's subscribe permissions to be 'my.service.req', got %q", subPerm)
}
// We want allow_responses to essentially set deny all, or allow none in this case.
if svcb.Permissions.Publish == nil {
t.Fatalf("Expected Service B's publish permissions to be non-nil")
}
if len(svcb.Permissions.Publish.Allow) != 0 {
t.Fatalf("Expected Service B's publish permissions to have no elements, got %d",
len(svcb.Permissions.Publish.Allow))
}
// We should have a ResponsePermission present with default values.
if svcb.Permissions.Response == nil {
t.Fatalf("Expected Service B's response permissions to be non-nil")
}
if svcb.Permissions.Response.MaxMsgs != 10 {
t.Fatalf("Expected Service B's response permissions of max msgs to be %d, got %d",
10, svcb.Permissions.Response.MaxMsgs,
)
}
if svcb.Permissions.Response.Expires != time.Minute {
t.Fatalf("Expected Service B's response permissions of expiration to be %v, got %v",
time.Minute, svcb.Permissions.Response.Expires,
)
}
}
@@ -655,13 +739,13 @@ func TestAuthorizationConfig(t *testing.T) {
func TestNewStyleAuthorizationConfig(t *testing.T) {
opts, err := ProcessConfigFile("./configs/new_style_authorization.conf")
if err != nil {
t.Fatalf("Received an error reading config file: %v\n", err)
t.Fatalf("Received an error reading config file: %v", err)
}
setBaselineOptions(opts)
lu := len(opts.Users)
if lu != 2 {
t.Fatalf("Expected 2 users, got %d\n", lu)
t.Fatalf("Expected 2 users, got %d", lu)
}
// Build a map
mu := make(map[string]*User)
@@ -671,99 +755,99 @@ func TestNewStyleAuthorizationConfig(t *testing.T) {
// Alice
alice, ok := mu["alice"]
if !ok {
t.Fatalf("Expected to see user Alice\n")
t.Fatalf("Expected to see user Alice")
}
if alice.Permissions == nil {
t.Fatalf("Expected Alice's permissions to be non-nil\n")
t.Fatalf("Expected Alice's permissions to be non-nil")
}
if alice.Permissions.Publish == nil {
t.Fatalf("Expected Alice's publish permissions to be non-nil\n")
t.Fatalf("Expected Alice's publish permissions to be non-nil")
}
if len(alice.Permissions.Publish.Allow) != 3 {
t.Fatalf("Expected Alice's allowed publish permissions to have 3 elements, got %d\n",
t.Fatalf("Expected Alice's allowed publish permissions to have 3 elements, got %d",
len(alice.Permissions.Publish.Allow))
}
pubPerm := alice.Permissions.Publish.Allow[0]
if pubPerm != "foo" {
t.Fatalf("Expected Alice's first allowed publish permission to be 'foo', got %q\n", pubPerm)
t.Fatalf("Expected Alice's first allowed publish permission to be 'foo', got %q", pubPerm)
}
pubPerm = alice.Permissions.Publish.Allow[1]
if pubPerm != "bar" {
t.Fatalf("Expected Alice's second allowed publish permission to be 'bar', got %q\n", pubPerm)
t.Fatalf("Expected Alice's second allowed publish permission to be 'bar', got %q", pubPerm)
}
pubPerm = alice.Permissions.Publish.Allow[2]
if pubPerm != "baz" {
t.Fatalf("Expected Alice's third allowed publish permission to be 'baz', got %q\n", pubPerm)
t.Fatalf("Expected Alice's third allowed publish permission to be 'baz', got %q", pubPerm)
}
if len(alice.Permissions.Publish.Deny) != 0 {
t.Fatalf("Expected Alice's denied publish permissions to have 0 elements, got %d\n",
t.Fatalf("Expected Alice's denied publish permissions to have 0 elements, got %d",
len(alice.Permissions.Publish.Deny))
}
if alice.Permissions.Subscribe == nil {
t.Fatalf("Expected Alice's subscribe permissions to be non-nil\n")
t.Fatalf("Expected Alice's subscribe permissions to be non-nil")
}
if len(alice.Permissions.Subscribe.Allow) != 0 {
t.Fatalf("Expected Alice's allowed subscribe permissions to have 0 elements, got %d\n",
t.Fatalf("Expected Alice's allowed subscribe permissions to have 0 elements, got %d",
len(alice.Permissions.Subscribe.Allow))
}
if len(alice.Permissions.Subscribe.Deny) != 1 {
t.Fatalf("Expected Alice's denied subscribe permissions to have 1 element, got %d\n",
t.Fatalf("Expected Alice's denied subscribe permissions to have 1 element, got %d",
len(alice.Permissions.Subscribe.Deny))
}
subPerm := alice.Permissions.Subscribe.Deny[0]
if subPerm != "$SYSTEM.>" {
t.Fatalf("Expected Alice's only denied subscribe permission to be '$SYSTEM.>', got %q\n", subPerm)
t.Fatalf("Expected Alice's only denied subscribe permission to be '$SYSTEM.>', got %q", subPerm)
}
// Bob
bob, ok := mu["bob"]
if !ok {
t.Fatalf("Expected to see user Bob\n")
t.Fatalf("Expected to see user Bob")
}
if bob.Permissions == nil {
t.Fatalf("Expected Bob's permissions to be non-nil\n")
t.Fatalf("Expected Bob's permissions to be non-nil")
}
if bob.Permissions.Publish == nil {
t.Fatalf("Expected Bobs's publish permissions to be non-nil\n")
t.Fatalf("Expected Bobs's publish permissions to be non-nil")
}
if len(bob.Permissions.Publish.Allow) != 1 {
t.Fatalf("Expected Bob's allowed publish permissions to have 1 element, got %d\n",
t.Fatalf("Expected Bob's allowed publish permissions to have 1 element, got %d",
len(bob.Permissions.Publish.Allow))
}
pubPerm = bob.Permissions.Publish.Allow[0]
if pubPerm != "$SYSTEM.>" {
t.Fatalf("Expected Bob's first allowed publish permission to be '$SYSTEM.>', got %q\n", pubPerm)
t.Fatalf("Expected Bob's first allowed publish permission to be '$SYSTEM.>', got %q", pubPerm)
}
if len(bob.Permissions.Publish.Deny) != 0 {
t.Fatalf("Expected Bob's denied publish permissions to have 0 elements, got %d\n",
t.Fatalf("Expected Bob's denied publish permissions to have 0 elements, got %d",
len(bob.Permissions.Publish.Deny))
}
if bob.Permissions.Subscribe == nil {
t.Fatalf("Expected Bob's subscribe permissions to be non-nil\n")
t.Fatalf("Expected Bob's subscribe permissions to be non-nil")
}
if len(bob.Permissions.Subscribe.Allow) != 0 {
t.Fatalf("Expected Bob's allowed subscribe permissions to have 0 elements, got %d\n",
t.Fatalf("Expected Bob's allowed subscribe permissions to have 0 elements, got %d",
len(bob.Permissions.Subscribe.Allow))
}
if len(bob.Permissions.Subscribe.Deny) != 3 {
t.Fatalf("Expected Bobs's denied subscribe permissions to have 3 elements, got %d\n",
t.Fatalf("Expected Bobs's denied subscribe permissions to have 3 elements, got %d",
len(bob.Permissions.Subscribe.Deny))
}
subPerm = bob.Permissions.Subscribe.Deny[0]
if subPerm != "foo" {
t.Fatalf("Expected Bobs's first denied subscribe permission to be 'foo', got %q\n", subPerm)
t.Fatalf("Expected Bobs's first denied subscribe permission to be 'foo', got %q", subPerm)
}
subPerm = bob.Permissions.Subscribe.Deny[1]
if subPerm != "bar" {
t.Fatalf("Expected Bobs's second denied subscribe permission to be 'bar', got %q\n", subPerm)
t.Fatalf("Expected Bobs's second denied subscribe permission to be 'bar', got %q", subPerm)
}
subPerm = bob.Permissions.Subscribe.Deny[2]
if subPerm != "baz" {
t.Fatalf("Expected Bobs's third denied subscribe permission to be 'baz', got %q\n", subPerm)
t.Fatalf("Expected Bobs's third denied subscribe permission to be 'baz', got %q", subPerm)
}
}
@@ -816,7 +900,7 @@ func TestNkeyUsersWithPermsConfig(t *testing.T) {
t.Fatal("Expected to have publish permissions")
}
if nk.Permissions.Publish.Allow[0] != "$SYSTEM.>" {
t.Fatalf("Expected publish to allow \"$SYSTEM.>\", but got %v\n", nk.Permissions.Publish.Allow[0])
t.Fatalf("Expected publish to allow \"$SYSTEM.>\", but got %v", nk.Permissions.Publish.Allow[0])
}
if nk.Permissions.Subscribe == nil {
t.Fatal("Expected to have subscribe permissions")
@@ -909,7 +993,7 @@ func TestTokenWithUsers(t *testing.T) {
func TestParseWriteDeadline(t *testing.T) {
confFile := "test.conf"
defer os.Remove(confFile)
if err := ioutil.WriteFile(confFile, []byte("write_deadline: \"1x\"\n"), 0666); err != nil {
if err := ioutil.WriteFile(confFile, []byte("write_deadline: \"1x\""), 0666); err != nil {
t.Fatalf("Error writing config file: %v", err)
}
_, err := ProcessConfigFile(confFile)
@@ -920,7 +1004,7 @@ func TestParseWriteDeadline(t *testing.T) {
t.Fatalf("Expected error related to parsing, got %v", err)
}
os.Remove(confFile)
if err := ioutil.WriteFile(confFile, []byte("write_deadline: \"1s\"\n"), 0666); err != nil {
if err := ioutil.WriteFile(confFile, []byte("write_deadline: \"1s\""), 0666); err != nil {
t.Fatalf("Error writing config file: %v", err)
}
opts, err := ProcessConfigFile(confFile)
@@ -938,7 +1022,7 @@ func TestParseWriteDeadline(t *testing.T) {
os.Stdout = oldStdout
}()
os.Stdout = w
if err := ioutil.WriteFile(confFile, []byte("write_deadline: 2\n"), 0666); err != nil {
if err := ioutil.WriteFile(confFile, []byte("write_deadline: 2"), 0666); err != nil {
t.Fatalf("Error writing config file: %v", err)
}
opts, err = ProcessConfigFile(confFile)
@@ -1044,14 +1128,14 @@ func TestEmptyConfig(t *testing.T) {
func TestMalformedListenAddress(t *testing.T) {
opts, err := ProcessConfigFile("./configs/malformed_listen_address.conf")
if err == nil {
t.Fatalf("Expected an error reading config file: got %+v\n", opts)
t.Fatalf("Expected an error reading config file: got %+v", opts)
}
}
func TestMalformedClusterAddress(t *testing.T) {
opts, err := ProcessConfigFile("./configs/malformed_cluster_address.conf")
if err == nil {
t.Fatalf("Expected an error reading config file: got %+v\n", opts)
t.Fatalf("Expected an error reading config file: got %+v", opts)
}
}
@@ -1757,9 +1841,9 @@ func TestParsingGatewaysErrors(t *testing.T) {
}
_, err := ProcessConfigFile(file)
if err == nil {
t.Fatalf("Expected to fail, did not. Content:\n%s\n", test.content)
t.Fatalf("Expected to fail, did not. Content:\n%s", test.content)
} else if !strings.Contains(err.Error(), test.expectedErr) {
t.Fatalf("Expected error containing %q, got %q, for content:\n%s\n", test.expectedErr, err, test.content)
t.Fatalf("Expected error containing %q, got %q, for content:\n%s", test.expectedErr, err, test.content)
}
})
}

2
server/route.go
View File

@@ -627,7 +627,7 @@ func (s *Server) forwardNewRouteInfoToKnownServers(info *Info) {
func (c *client) canImport(subject string) bool {
// Use pubAllowed() since this checks Publish permissions which
// is what Import maps to.
return c.pubAllowed(subject)
return c.pubAllowedFullCheck(subject, false)
}
// canExport is whether or not we will accept a SUB from the remote for a given subject.

2
test/configs/authorization.conf
View File

@@ -16,5 +16,7 @@ authorization {
{user: ns, password: $PASS, permissions: $NEW_STYLE}
{user: ns-pub, password: $PASS, permissions: $NS_PUB}
{user: bench-deny, password: $PASS, permissions: $BENCH_DENY}
{user: svca, password: $PASS, permissions: $MY_SERVICE}
{user: svcb, password: $PASS, permissions: $MY_STREAM_SERVICE}
]
}

17
test/configs/auths.conf
View File

@@ -50,4 +50,19 @@ BENCH_DENY = {
allow = ["foo", "*"]
deny = "foo.bar"
}
}
}
# This is for services where you only want
# responses to reply subjects to be allowed.
MY_SERVICE = {
subscribe = "my.service.req"
publish_allow_responses: true
}
# This is a more detailed example where responses
# could be streams and you want to set the TTL
# and maximum allowed.
MY_STREAM_SERVICE = {
subscribe = "my.service.req"
allow_responses = {max: 10, ttl: "10ms"}
}

95
test/user_authorization_test.go
View File

@@ -1,4 +1,4 @@
// Copyright 2016-2018 The NATS Authors
// Copyright 2016-2019 The NATS Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
@@ -16,6 +16,7 @@ package test
import (
"regexp"
"testing"
"time"
)
const DefaultPass = "foo"
@@ -221,3 +222,95 @@ func TestUserAuthorizationProto(t *testing.T) {
sendProto(t, c, "SUB SYS.bar 5\r\n")
expectResult(t, c, errRe)
}
func TestUserAuthorizationAllowResponses(t *testing.T) {
srv, opts := RunServerWithConfig("./configs/authorization.conf")
defer srv.Shutdown()
// Alice can do anything, so she will be our requestor
rc := createClientConn(t, opts.Host, opts.Port)
defer rc.Close()
expectAuthRequired(t, rc)
doAuthConnect(t, rc, "", "alice", DefaultPass)
expectResult(t, rc, okRe)
// MY_SERVICE can subscribe to a single request subject but can
// respond to any reply subject that it receives, but only
// for one response.
c := createClientConn(t, opts.Host, opts.Port)
defer c.Close()
expectAuthRequired(t, c)
doAuthConnect(t, c, "", "svca", DefaultPass)
expectResult(t, c, okRe)
sendProto(t, c, "SUB my.service.req 1\r\n")
expectResult(t, c, okRe)
sendProto(t, rc, "PUB my.service.req resp.bar.22 2\r\nok\r\n")
expectResult(t, rc, okRe)
matches := msgRe.FindAllSubmatch(expectResult(t, c, msgRe), -1)
checkMsg(t, matches[0], "my.service.req", "1", "resp.bar.22", "2", "ok")
// This should be allowed
sendProto(t, c, "PUB resp.bar.22 2\r\nok\r\n")
expectResult(t, c, okRe)
// This should not be allowed
sendProto(t, c, "PUB resp.bar.33 2\r\nok\r\n")
expectResult(t, c, errRe)
// This should also not be allowed now since we already sent a response and max is 1.
sendProto(t, c, "PUB resp.bar.22 2\r\nok\r\n")
expectResult(t, c, errRe)
c.Close() // from MY_SERVICE
// MY_STREAM_SERVICE can subscribe to a single request subject but can
// respond to any reply subject that it receives, and send up to 10 responses.
// Each permission for a response can last up to 10ms.
c = createClientConn(t, opts.Host, opts.Port)
defer c.Close()
expectAuthRequired(t, c)
doAuthConnect(t, c, "", "svcb", DefaultPass)
expectResult(t, c, okRe)
sendProto(t, c, "SUB my.service.req 1\r\n")
expectResult(t, c, okRe)
// Same rules as above.
sendProto(t, rc, "PUB my.service.req resp.bar.22 2\r\nok\r\n")
expectResult(t, rc, okRe)
matches = msgRe.FindAllSubmatch(expectResult(t, c, msgRe), -1)
checkMsg(t, matches[0], "my.service.req", "1", "resp.bar.22", "2", "ok")
// This should be allowed
sendProto(t, c, "PUB resp.bar.22 2\r\nok\r\n")
expectResult(t, c, okRe)
// This should not be allowed
sendProto(t, c, "PUB resp.bar.33 2\r\nok\r\n")
expectResult(t, c, errRe)
// We should be able to send 9 more here since we are allowed 10 total
for i := 0; i < 9; i++ {
sendProto(t, c, "PUB resp.bar.22 2\r\nok\r\n")
expectResult(t, c, okRe)
}
// Now this should fail since we already sent 10 responses.
sendProto(t, c, "PUB resp.bar.22 2\r\nok\r\n")
expectResult(t, c, errRe)
// Now test timeout.
sendProto(t, rc, "PUB my.service.req resp.bar.11 2\r\nok\r\n")
expectResult(t, rc, okRe)
matches = msgRe.FindAllSubmatch(expectResult(t, c, msgRe), -1)
checkMsg(t, matches[0], "my.service.req", "1", "resp.bar.11", "2", "ok")
time.Sleep(20 * time.Millisecond)
sendProto(t, c, "PUB resp.bar.11 2\r\nok\r\n")
expectResult(t, c, errRe)
}