Description: Apply changes from the trust branch which builds against the version of nkeys that we have in Debian. Origin: https://github.com/nats-io/gnatsd/tree/trust Author: Alexandre Viau Last-Update: 2018-11-21 From 50bfbe20ebc1fbe7ecb66d0909e14501919a595c Mon Sep 17 00:00:00 2001 From: aviau Date: Wed, 21 Nov 2018 11:13:36 -0500 Subject: [PATCH] trust branch --- .travis.yml | 1 + server/accounts.go | 479 +++++++++++++++-- server/accounts_test.go | 74 +-- server/auth.go | 90 +++- server/client.go | 125 +++-- server/client_test.go | 4 +- server/closed_conns_test.go | 2 +- server/const.go | 3 + server/errors.go | 15 +- server/jwt_test.go | 1013 +++++++++++++++++++++++++++++++++++ server/monitor.go | 2 + server/nkey.go | 3 +- server/nkey_test.go | 6 +- server/opts.go | 35 +- server/parser.go | 8 +- server/reload_test.go | 4 +- server/server.go | 178 +++++- server/trust_test.go | 123 +++++ server/util.go | 2 +- test/new_routes_test.go | 10 +- 20 files changed, 2018 insertions(+), 159 deletions(-) create mode 100644 server/jwt_test.go create mode 100644 server/trust_test.go diff --git a/.travis.yml b/.travis.yml index 5f6f2fa..5d087ec 100644 --- a/.travis.yml +++ b/.travis.yml @@ -5,6 +5,7 @@ go: install: - go get github.com/nats-io/go-nats - go get github.com/nats-io/nkeys +- go get github.com/nats-io/jwt - go get github.com/mattn/goveralls - go get github.com/wadey/gocovmerge - go get -u honnef.co/go/tools/cmd/megacheck diff --git a/server/accounts.go b/server/accounts.go index 6c52bb4..5760f16 100644 --- a/server/accounts.go +++ b/server/accounts.go @@ -14,10 +14,16 @@ package server import ( + "io/ioutil" + "net/http" + "net/url" + "reflect" "sort" "strings" "sync" "time" + + "github.com/nats-io/jwt" ) // For backwards compatibility, users who are not explicitly defined into an @@ -36,9 +42,13 @@ type rme struct { type Account struct { Name string Nkey string + Issuer string + claimJWT string + updated time.Time mu sync.RWMutex sl *Sublist - clients int + etmr *time.Timer + clients map[*client]*client rm map[string]*rme imports importMap exports exportMap @@ -46,22 +56,33 @@ type Account struct { maxnae int maxaettl time.Duration pruning bool + expired bool } // Import stream mapping struct type streamImport struct { - acc *Account - from string - prefix string + acc *Account + from string + prefix string + claim *jwt.Import + invalid bool } // Import service mapping struct type serviceImport struct { - acc *Account - from string - to string - ae bool - ts int64 + acc *Account + from string + to string + ae bool + ts int64 + claim *jwt.Import +} + +// exportAuth holds configured approvals or boolean indicating an +// auth token is required for import. +type exportAuth struct { + tokenReq bool + approved map[string]*Account } // importMap tracks the imported streams and services. @@ -72,15 +93,15 @@ type importMap struct { // exportMap tracks the exported streams and services. type exportMap struct { - streams map[string]map[string]*Account - services map[string]map[string]*Account + streams map[string]*exportAuth + services map[string]*exportAuth } // NumClients returns active number of clients for this account. func (a *Account) NumClients() int { a.mu.RLock() defer a.mu.RUnlock() - return a.clients + return len(a.clients) } // RoutedSubs returns how many subjects we would send across a route when first @@ -99,12 +120,11 @@ func (a *Account) TotalSubs() int { } // addClient keeps our accounting of active clients updated. -// Call in with client but just track total for now. // Returns previous total. func (a *Account) addClient(c *client) int { a.mu.Lock() - n := a.clients - a.clients++ + n := len(a.clients) + a.clients[c] = c a.mu.Unlock() return n } @@ -112,8 +132,8 @@ func (a *Account) addClient(c *client) int { // removeClient keeps our accounting of active clients updated. func (a *Account) removeClient(c *client) int { a.mu.Lock() - n := a.clients - a.clients-- + n := len(a.clients) + delete(a.clients, c) a.mu.Unlock() return n } @@ -126,16 +146,22 @@ func (a *Account) AddServiceExport(subject string, accounts []*Account) error { return ErrMissingAccount } if a.exports.services == nil { - a.exports.services = make(map[string]map[string]*Account) + a.exports.services = make(map[string]*exportAuth) } - ma := a.exports.services[subject] - if accounts != nil && ma == nil { - ma = make(map[string]*Account) - } - for _, a := range accounts { - ma[a.Name] = a + ea := a.exports.services[subject] + if accounts != nil && ea == nil { + ea = &exportAuth{} + // empty means auth required but will be import token. + if len(accounts) == 0 { + ea.tokenReq = true + } else { + ea.approved = make(map[string]*Account) + for _, acc := range accounts { + ea.approved[acc.Name] = acc + } + } } - a.exports.services[subject] = ma + a.exports.services[subject] = ea return nil } @@ -146,11 +172,7 @@ func (a *Account) numServiceRoutes() int { return len(a.imports.services) } -// AddServiceImport will add a route to an account to send published messages / requests -// to the destination account. From is the local subject to map, To is the -// subject that will appear on the destination account. Destination will need -// to have an import rule to allow access via addService. -func (a *Account) AddServiceImport(destination *Account, from, to string) error { +func (a *Account) AddServiceImportWithClaim(destination *Account, from, to string, imClaim *jwt.Import) error { if destination == nil { return ErrMissingAccount } @@ -162,11 +184,19 @@ func (a *Account) AddServiceImport(destination *Account, from, to string) error return ErrInvalidSubject } // First check to see if the account has authorized us to route to the "to" subject. - if !destination.checkServiceImportAuthorized(a, to) { + if !destination.checkServiceImportAuthorized(a, to, imClaim) { return ErrServiceImportAuthorization } - return a.addImplicitServiceImport(destination, from, to, false) + return a.addImplicitServiceImport(destination, from, to, false, imClaim) +} + +// AddServiceImport will add a route to an account to send published messages / requests +// to the destination account. From is the local subject to map, To is the +// subject that will appear on the destination account. Destination will need +// to have an import rule to allow access via addService. +func (a *Account) AddServiceImport(destination *Account, from, to string) error { + return a.AddServiceImportWithClaim(destination, from, to, nil) } // removeServiceImport will remove the route by subject. @@ -239,12 +269,12 @@ func (a *Account) autoExpireResponseMaps() []*serviceImport { // Add a route to connect from an implicit route created for a response to a request. // This does no checks and should be only called by the msg processing code. Use // addServiceImport from above if responding to user input or config, etc. -func (a *Account) addImplicitServiceImport(destination *Account, from, to string, autoexpire bool) error { +func (a *Account) addImplicitServiceImport(destination *Account, from, to string, autoexpire bool, claim *jwt.Import) error { a.mu.Lock() if a.imports.services == nil { a.imports.services = make(map[string]*serviceImport) } - si := &serviceImport{destination, from, to, autoexpire, 0} + si := &serviceImport{destination, from, to, autoexpire, 0, claim} a.imports.services[from] = si if autoexpire { a.nae++ @@ -300,14 +330,14 @@ func (a *Account) pruneAutoExpireResponseMaps() { } } -// AddStreamImport will add in the stream import from a specific account. -func (a *Account) AddStreamImport(account *Account, from, prefix string) error { +// AddStreamImport will add in the stream import from a specific account with optional token. +func (a *Account) AddStreamImportWithClaim(account *Account, from, prefix string, imClaim *jwt.Import) error { if account == nil { return ErrMissingAccount } // First check to see if the account has authorized export of the subject. - if !account.checkStreamImportAuthorized(a, from) { + if !account.checkStreamImportAuthorized(a, from, imClaim) { return ErrStreamImportAuthorization } @@ -320,10 +350,15 @@ func (a *Account) AddStreamImport(account *Account, from, prefix string) error { prefix = prefix + string(btsep) } // TODO(dlc) - collisions, etc. - a.imports.streams[from] = &streamImport{account, from, prefix} + a.imports.streams[from] = &streamImport{account, from, prefix, imClaim, false} return nil } +// AddStreamImport will add in the stream import from a specific account. +func (a *Account) AddStreamImport(account *Account, from, prefix string) error { + return a.AddStreamImportWithClaim(account, from, prefix, nil) +} + // IsPublicExport is a placeholder to denote public export. var IsPublicExport = []*Account(nil) @@ -336,38 +371,51 @@ func (a *Account) AddStreamExport(subject string, accounts []*Account) error { return ErrMissingAccount } if a.exports.streams == nil { - a.exports.streams = make(map[string]map[string]*Account) + a.exports.streams = make(map[string]*exportAuth) } - var ma map[string]*Account - for _, aa := range accounts { - if ma == nil { - ma = make(map[string]*Account, len(accounts)) + ea := a.exports.services[subject] + if accounts != nil && ea == nil { + ea = &exportAuth{} + // empty means auth required but will be import token. + if len(accounts) == 0 { + ea.tokenReq = true + } else { + ea.approved = make(map[string]*Account) } - ma[aa.Name] = aa } - a.exports.streams[subject] = ma + for _, acc := range accounts { + ea.approved[acc.Name] = acc + } + a.exports.streams[subject] = ea return nil } // Check if another account is authorized to import from us. -func (a *Account) checkStreamImportAuthorized(account *Account, subject string) bool { +func (a *Account) checkStreamImportAuthorized(account *Account, subject string, imClaim *jwt.Import) bool { // Find the subject in the exports list. a.mu.RLock() defer a.mu.RUnlock() + return a.checkStreamImportAuthorizedNoLock(account, subject, imClaim) +} +func (a *Account) checkStreamImportAuthorizedNoLock(account *Account, subject string, imClaim *jwt.Import) bool { if a.exports.streams == nil || !IsValidSubject(subject) { return false } // Check direct match of subject first - am, ok := a.exports.streams[subject] + ea, ok := a.exports.streams[subject] if ok { - // if am is nil that denotes a public export - if am == nil { + // if ea is nil that denotes a public export + if ea == nil { return true } + // Check if token required + if ea != nil && ea.tokenReq { + return a.checkActivation(account, imClaim, true) + } // If we have a matching account we are authorized - _, ok := am[account.Name] + _, ok := ea.approved[account.Name] return ok } // ok if we are here we did not match directly so we need to test each one. @@ -375,18 +423,112 @@ func (a *Account) checkStreamImportAuthorized(account *Account, subject string) // has to be a true subset of the import claim. We already checked for // exact matches above. tokens := strings.Split(subject, tsep) - for subj, am := range a.exports.streams { + for subj, ea := range a.exports.streams { if isSubsetMatch(tokens, subj) { - if am == nil { + if ea == nil || ea.approved == nil { return true } - _, ok := am[account.Name] + // Check if token required + if ea != nil && ea.tokenReq { + return a.checkActivation(account, imClaim, true) + } + _, ok := ea.approved[account.Name] return ok } } return false } +// Will fetch the activation token for an import. +func fetchActivation(url string) string { + // FIXME(dlc) - Make configurable. + c := &http.Client{Timeout: 2 * time.Second} + resp, err := c.Get(url) + if err != nil || resp == nil { + return "" + } + defer resp.Body.Close() + body, err := ioutil.ReadAll(resp.Body) + if err != nil { + return "" + } + return string(body) +} + +// Fires for expired activation tokens. We could track this with timers etc. +// Instead we just re-analyze where we are and if we need to act. +func (a *Account) activationExpired(subject string) { + a.mu.RLock() + if a.expired || a.imports.streams == nil { + a.mu.RUnlock() + return + } + // FIXME(dlc) - check services too? + si := a.imports.streams[subject] + a.mu.RUnlock() + + if si == nil || si.invalid { + return + } + if si.acc.checkActivation(a, si.claim, false) { + // The token has been updated most likely and we are good to go. + return + } + + a.mu.Lock() + si.invalid = true + clients := make([]*client, 0, len(a.clients)) + for _, c := range a.clients { + clients = append(clients, c) + } + awcsti := map[string]struct{}{a.Name: struct{}{}} + a.mu.Unlock() + for _, c := range clients { + c.processSubsOnConfigReload(awcsti) + } +} + +// checkActivation will check the activation token for validity. +func (a *Account) checkActivation(acc *Account, claim *jwt.Import, expTimer bool) bool { + if claim == nil || claim.Token == "" { + return false + } + // Create a quick clone so we can inline Token JWT. + clone := *claim + + // We grab the token from a URL by hand here since we need expiration etc. + if url, err := url.Parse(clone.Token); err == nil && url.Scheme != "" { + clone.Token = fetchActivation(url.String()) + } + vr := jwt.CreateValidationResults() + clone.Validate(a.Name, vr) + if vr.IsBlocking(true) { + return false + } + act, err := jwt.DecodeActivationClaims(clone.Token) + if err != nil { + return false + } + vr = jwt.CreateValidationResults() + act.Validate(vr) + if vr.IsBlocking(true) { + return false + } + if act.Expires != 0 { + tn := time.Now().Unix() + if act.Expires <= tn { + return false + } + if expTimer && len(act.Exports) > 0 { + expiresAt := time.Duration(act.Expires - tn) + time.AfterFunc(expiresAt*time.Second, func() { + acc.activationExpired(string(act.Exports[0].Subject)) + }) + } + } + return true +} + // Returns true if `a` and `b` stream imports are the same. Note that the // check is done with the account's name, not the pointer. This is used // during config reload where we are comparing current and new config @@ -409,8 +551,24 @@ func (a *Account) checkStreamImportsEqual(b *Account) bool { return true } +func (a *Account) checkStreamExportsEqual(b *Account) bool { + if len(a.exports.streams) != len(b.exports.streams) { + return false + } + for subj, aea := range a.exports.streams { + bea, ok := b.exports.streams[subj] + if !ok { + return false + } + if !reflect.DeepEqual(aea, bea) { + return false + } + } + return true +} + // Check if another account is authorized to route requests to this service. -func (a *Account) checkServiceImportAuthorized(account *Account, subject string) bool { +func (a *Account) checkServiceImportAuthorized(account *Account, subject string, imClaim *jwt.Import) bool { // Find the subject in the services list. a.mu.RLock() defer a.mu.RUnlock() @@ -419,15 +577,222 @@ func (a *Account) checkServiceImportAuthorized(account *Account, subject string) return false } // These are always literal subjects so just lookup. - am, ok := a.exports.services[subject] + ae, ok := a.exports.services[subject] if !ok { return false } + + if ae != nil && ae.tokenReq { + return a.checkActivation(account, imClaim, false) + } + // Check to see if we are public or if we need to search for the account. - if am == nil { + if ae == nil || ae.approved == nil { return true } // Check that we allow this account. - _, ok = am[account.Name] + _, ok = ae.approved[account.Name] return ok } + +// IsExpired returns expiration status. +func (a *Account) IsExpired() bool { + a.mu.RLock() + defer a.mu.RUnlock() + return a.expired +} + +// Called when an account has expired. +func (a *Account) expiredTimeout() { + // Collect the clients. + a.mu.Lock() + a.expired = true + a.mu.Unlock() + + cs := make([]*client, 0, len(a.clients)) + a.mu.RLock() + for c := range a.clients { + cs = append(cs, c) + } + a.mu.RUnlock() + + for _, c := range cs { + c.accountAuthExpired() + } +} + +// Sets the expiration timer for an account JWT that has it set. +func (a *Account) setExpirationTimer(d time.Duration) { + a.etmr = time.AfterFunc(d, a.expiredTimeout) +} + +// Lock should be held +func (a *Account) clearExpirationTimer() bool { + if a.etmr == nil { + return true + } + stopped := a.etmr.Stop() + a.etmr = nil + return stopped +} + +func (a *Account) checkExpiration(claims *jwt.ClaimsData) { + a.clearExpirationTimer() + if claims.Expires == 0 { + a.expired = false + return + } + tn := time.Now().Unix() + if claims.Expires <= tn { + a.expired = true + return + } + expiresAt := time.Duration(claims.Expires - tn) + a.setExpirationTimer(expiresAt * time.Second) + a.expired = false +} + +// Placeholder for signaling token auth required. +var tokenAuthReq = []*Account{} + +func authAccounts(tokenReq bool) []*Account { + if tokenReq { + return tokenAuthReq + } + return nil +} + +// UpdateAccountClaims will update and existing account with new claims. +// This will replace any exports or imports previously defined. +func (s *Server) UpdateAccountClaims(a *Account, ac *jwt.AccountClaims) { + if a == nil { + return + } + a.checkExpiration(ac.Claims()) + + // Clone to update + old := &Account{Name: a.Name, imports: a.imports, exports: a.exports} + + // Reset exports and imports here. + a.exports = exportMap{} + a.imports = importMap{} + + for _, e := range ac.Exports { + switch e.Type { + case jwt.Stream: + if err := a.AddStreamExport(string(e.Subject), authAccounts(e.TokenReq)); err != nil { + s.Debugf("Error adding stream export to account [%s]: %v", a.Name, err.Error()) + } + case jwt.Service: + if err := a.AddServiceExport(string(e.Subject), authAccounts(e.TokenReq)); err != nil { + s.Debugf("Error adding service export to account [%s]: %v", a.Name, err.Error()) + } + } + } + for _, i := range ac.Imports { + acc := s.accounts[i.Account] + if acc == nil { + if acc = s.FetchAccount(i.Account); acc == nil { + s.Debugf("Can't locate account [%s] for import of [%v] %s", i.Account, i.Subject, i.Type) + continue + } + } + switch i.Type { + case jwt.Stream: + a.AddStreamImportWithClaim(acc, string(i.Subject), string(i.To), i) + case jwt.Service: + a.AddServiceImportWithClaim(acc, string(i.Subject), string(i.To), i) + } + } + // Now let's apply any needed changes from import/export changes. + if !a.checkStreamImportsEqual(old) { + awcsti := map[string]struct{}{a.Name: struct{}{}} + a.mu.RLock() + clients := make([]*client, 0, len(a.clients)) + for _, c := range a.clients { + clients = append(clients, c) + } + a.mu.RUnlock() + for _, c := range clients { + c.processSubsOnConfigReload(awcsti) + } + } + // Now check if exports have changed. + if !a.checkStreamExportsEqual(old) { + clients := make([]*client, 0, 16) + // We need to check all accounts that have an import claim from this account. + awcsti := map[string]struct{}{} + for _, acc := range s.accounts { + acc.mu.Lock() + for _, im := range acc.imports.streams { + if im != nil && im.acc.Name == a.Name { + // Check for if we are still authorized for an import. + im.invalid = !a.checkStreamImportAuthorizedNoLock(im.acc, im.from, im.claim) + awcsti[acc.Name] = struct{}{} + for _, c := range acc.clients { + clients = append(clients, c) + } + break + } + } + acc.mu.Unlock() + } + // Now walk clients. + for _, c := range clients { + c.processSubsOnConfigReload(awcsti) + } + } + + // FIXME(dlc) - Limits etc.. +} + +// Helper to build an internal account structure from a jwt.AccountClaims. +func (s *Server) buildInternalAccount(ac *jwt.AccountClaims) *Account { + acc := &Account{Name: ac.Subject, Issuer: ac.Issuer} + s.UpdateAccountClaims(acc, ac) + return acc +} + +// Helper to build internal NKeyUser. +func buildInternalNkeyUser(uc *jwt.UserClaims, acc *Account) *NkeyUser { + nu := &NkeyUser{Nkey: uc.Subject, Account: acc} + + // Now check for permissions. + var p *Permissions + + if len(uc.Pub.Allow) > 0 || len(uc.Pub.Deny) > 0 { + if p == nil { + p = &Permissions{} + } + p.Publish = &SubjectPermission{} + p.Publish.Allow = uc.Pub.Allow + p.Publish.Deny = uc.Pub.Deny + } + if len(uc.Sub.Allow) > 0 || len(uc.Sub.Deny) > 0 { + if p == nil { + p = &Permissions{} + } + p.Subscribe = &SubjectPermission{} + p.Subscribe.Allow = uc.Sub.Allow + p.Subscribe.Deny = uc.Sub.Deny + } + nu.Permissions = p + return nu +} + +// AccountResolver interface. This is to fetch Account JWTs by public nkeys +type AccountResolver interface { + Fetch(pub string) (string, error) +} + +// Mostly for testing. +type memAccResolver struct { + sync.Map +} + +func (m *memAccResolver) Fetch(pub string) (string, error) { + if j, ok := m.Load(pub); ok { + return j.(string), nil + } + return "", ErrMissingAccount +} diff --git a/server/accounts_test.go b/server/accounts_test.go index 3d6fed1..43a5a1f 100644 --- a/server/accounts_test.go +++ b/server/accounts_test.go @@ -527,60 +527,60 @@ func TestImportExportConfigFailures(t *testing.T) { func TestImportAuthorized(t *testing.T) { _, foo, bar := simpleAccountServer(t) - checkBool(foo.checkStreamImportAuthorized(bar, "foo"), false, t) - checkBool(foo.checkStreamImportAuthorized(bar, "*"), false, t) - checkBool(foo.checkStreamImportAuthorized(bar, ">"), false, t) - checkBool(foo.checkStreamImportAuthorized(bar, "foo.*"), false, t) - checkBool(foo.checkStreamImportAuthorized(bar, "foo.>"), false, t) + checkBool(foo.checkStreamImportAuthorized(bar, "foo", nil), false, t) + checkBool(foo.checkStreamImportAuthorized(bar, "*", nil), false, t) + checkBool(foo.checkStreamImportAuthorized(bar, ">", nil), false, t) + checkBool(foo.checkStreamImportAuthorized(bar, "foo.*", nil), false, t) + checkBool(foo.checkStreamImportAuthorized(bar, "foo.>", nil), false, t) foo.AddStreamExport("foo", IsPublicExport) - checkBool(foo.checkStreamImportAuthorized(bar, "foo"), true, t) - checkBool(foo.checkStreamImportAuthorized(bar, "bar"), false, t) - checkBool(foo.checkStreamImportAuthorized(bar, "*"), false, t) + checkBool(foo.checkStreamImportAuthorized(bar, "foo", nil), true, t) + checkBool(foo.checkStreamImportAuthorized(bar, "bar", nil), false, t) + checkBool(foo.checkStreamImportAuthorized(bar, "*", nil), false, t) foo.AddStreamExport("*", []*Account{bar}) - checkBool(foo.checkStreamImportAuthorized(bar, "foo"), true, t) - checkBool(foo.checkStreamImportAuthorized(bar, "bar"), true, t) - checkBool(foo.checkStreamImportAuthorized(bar, "baz"), true, t) - checkBool(foo.checkStreamImportAuthorized(bar, "foo.bar"), false, t) - checkBool(foo.checkStreamImportAuthorized(bar, ">"), false, t) - checkBool(foo.checkStreamImportAuthorized(bar, "*"), true, t) - checkBool(foo.checkStreamImportAuthorized(bar, "foo.*"), false, t) - checkBool(foo.checkStreamImportAuthorized(bar, "*.*"), false, t) - checkBool(foo.checkStreamImportAuthorized(bar, "*.>"), false, t) + checkBool(foo.checkStreamImportAuthorized(bar, "foo", nil), true, t) + checkBool(foo.checkStreamImportAuthorized(bar, "bar", nil), true, t) + checkBool(foo.checkStreamImportAuthorized(bar, "baz", nil), true, t) + checkBool(foo.checkStreamImportAuthorized(bar, "foo.bar", nil), false, t) + checkBool(foo.checkStreamImportAuthorized(bar, ">", nil), false, t) + checkBool(foo.checkStreamImportAuthorized(bar, "*", nil), true, t) + checkBool(foo.checkStreamImportAuthorized(bar, "foo.*", nil), false, t) + checkBool(foo.checkStreamImportAuthorized(bar, "*.*", nil), false, t) + checkBool(foo.checkStreamImportAuthorized(bar, "*.>", nil), false, t) // Reset and test '>' public export _, foo, bar = simpleAccountServer(t) foo.AddStreamExport(">", nil) // Everything should work. - checkBool(foo.checkStreamImportAuthorized(bar, "foo"), true, t) - checkBool(foo.checkStreamImportAuthorized(bar, "bar"), true, t) - checkBool(foo.checkStreamImportAuthorized(bar, "baz"), true, t) - checkBool(foo.checkStreamImportAuthorized(bar, "foo.bar"), true, t) - checkBool(foo.checkStreamImportAuthorized(bar, ">"), true, t) - checkBool(foo.checkStreamImportAuthorized(bar, "*"), true, t) - checkBool(foo.checkStreamImportAuthorized(bar, "foo.*"), true, t) - checkBool(foo.checkStreamImportAuthorized(bar, "*.*"), true, t) - checkBool(foo.checkStreamImportAuthorized(bar, "*.>"), true, t) + checkBool(foo.checkStreamImportAuthorized(bar, "foo", nil), true, t) + checkBool(foo.checkStreamImportAuthorized(bar, "bar", nil), true, t) + checkBool(foo.checkStreamImportAuthorized(bar, "baz", nil), true, t) + checkBool(foo.checkStreamImportAuthorized(bar, "foo.bar", nil), true, t) + checkBool(foo.checkStreamImportAuthorized(bar, ">", nil), true, t) + checkBool(foo.checkStreamImportAuthorized(bar, "*", nil), true, t) + checkBool(foo.checkStreamImportAuthorized(bar, "foo.*", nil), true, t) + checkBool(foo.checkStreamImportAuthorized(bar, "*.*", nil), true, t) + checkBool(foo.checkStreamImportAuthorized(bar, "*.>", nil), true, t) // Reset and test pwc and fwc s, foo, bar := simpleAccountServer(t) foo.AddStreamExport("foo.*.baz.>", []*Account{bar}) - checkBool(foo.checkStreamImportAuthorized(bar, "foo.bar.baz.1"), true, t) - checkBool(foo.checkStreamImportAuthorized(bar, "foo.bar.baz.*"), true, t) - checkBool(foo.checkStreamImportAuthorized(bar, "foo.*.baz.1.1"), true, t) - checkBool(foo.checkStreamImportAuthorized(bar, "foo.22.baz.22"), true, t) - checkBool(foo.checkStreamImportAuthorized(bar, "foo.bar.baz"), false, t) - checkBool(foo.checkStreamImportAuthorized(bar, ""), false, t) - checkBool(foo.checkStreamImportAuthorized(bar, "foo.bar.*.*"), false, t) + checkBool(foo.checkStreamImportAuthorized(bar, "foo.bar.baz.1", nil), true, t) + checkBool(foo.checkStreamImportAuthorized(bar, "foo.bar.baz.*", nil), true, t) + checkBool(foo.checkStreamImportAuthorized(bar, "foo.*.baz.1.1", nil), true, t) + checkBool(foo.checkStreamImportAuthorized(bar, "foo.22.baz.22", nil), true, t) + checkBool(foo.checkStreamImportAuthorized(bar, "foo.bar.baz", nil), false, t) + checkBool(foo.checkStreamImportAuthorized(bar, "", nil), false, t) + checkBool(foo.checkStreamImportAuthorized(bar, "foo.bar.*.*", nil), false, t) // Make sure we match the account as well fb, _ := s.RegisterAccount("foobar") bz, _ := s.RegisterAccount("baz") - checkBool(foo.checkStreamImportAuthorized(fb, "foo.bar.baz.1"), false, t) - checkBool(foo.checkStreamImportAuthorized(bz, "foo.bar.baz.1"), false, t) + checkBool(foo.checkStreamImportAuthorized(fb, "foo.bar.baz.1", nil), false, t) + checkBool(foo.checkStreamImportAuthorized(bz, "foo.bar.baz.1", nil), false, t) } func TestSimpleMapping(t *testing.T) { @@ -1135,7 +1135,7 @@ func TestAccountMapsUsers(t *testing.T) { } // Now test nkeys as well. - kp, _ := nkeys.FromSeed(seed1) + kp, _ := nkeys.FromSeed([]byte(seed1)) pubKey, _ := kp.PublicKey() c, cr, l := newClientForServer(s) @@ -1166,7 +1166,7 @@ func TestAccountMapsUsers(t *testing.T) { } // Now nats account nkey user. - kp, _ = nkeys.FromSeed(seed2) + kp, _ = nkeys.FromSeed([]byte(seed2)) pubKey, _ = kp.PublicKey() c, cr, l = newClientForServer(s) diff --git a/server/auth.go b/server/auth.go index 44464d2..963986b 100644 --- a/server/auth.go +++ b/server/auth.go @@ -18,6 +18,7 @@ import ( "encoding/base64" "strings" + "github.com/nats-io/jwt" "github.com/nats-io/nkeys" "golang.org/x/crypto/bcrypt" ) @@ -181,6 +182,8 @@ func (s *Server) configureAuthorization() { // This just checks and sets up the user map if we have multiple users. if opts.CustomClientAuthentication != nil { s.info.AuthRequired = true + } else if len(s.trustedNkeys) > 0 { + s.info.AuthRequired = true } else if opts.Nkeys != nil || opts.Users != nil { // Support both at the same time. if opts.Nkeys != nil { @@ -205,9 +208,9 @@ func (s *Server) configureAuthorization() { } } -// checkAuthorization will check authorization based on client type and +// checkAuthentication will check based on client type and // return boolean indicating if client is authorized. -func (s *Server) checkAuthorization(c *client) bool { +func (s *Server) checkAuthentication(c *client) bool { switch c.typ { case CLIENT: return s.isClientAuthorized(c) @@ -229,14 +232,20 @@ func (s *Server) isClientAuthorized(c *client) bool { password := s.opts.Password s.optsMu.RUnlock() - // Check custom auth first, then nkeys, then multiple users, then token, then single user/pass. + // Check custom auth first, then jwts, then nkeys, then multiple users, then token, then single user/pass. if customClientAuthentication != nil { return customClientAuthentication.Check(c) } - var nkey *NkeyUser - var user *User - var ok bool + // Grab under lock but process after. + var ( + nkey *NkeyUser + juc *jwt.UserClaims + acc *Account + user *User + ok bool + err error + ) s.mu.Lock() authRequired := s.info.AuthRequired @@ -246,6 +255,30 @@ func (s *Server) isClientAuthorized(c *client) bool { return true } + // Check if we have trustedNkeys defined in the server. If so we require a user jwt. + if s.trustedNkeys != nil { + if c.opts.JWT == "" { + s.mu.Unlock() + c.Debugf("Authentication requires a user JWT") + return false + } + // So we have a valid user jwt here. + juc, err = jwt.DecodeUserClaims(c.opts.JWT) + if err != nil { + // Should we debug log here? + s.mu.Unlock() + c.Debugf("User JWT not valid: %v", err) + return false + } + vr := jwt.CreateValidationResults() + juc.Validate(vr) + if vr.IsBlocking(true) { + s.mu.Unlock() + c.Debugf("User JWT no longer valid: %+v", vr) + return false + } + } + // Check if we have nkeys or users for client. hasNkeys := s.nkeys != nil hasUsers := s.users != nil @@ -264,7 +297,48 @@ func (s *Server) isClientAuthorized(c *client) bool { } s.mu.Unlock() - // Verify the signature against the nonce. + // If we have a jwt and a userClaim, make sure we have the Account, etc associated. + // We need to look up the account. This will use an account resolver if one is present. + if juc != nil { + if acc = s.LookupAccount(juc.Issuer); acc == nil { + c.Debugf("Account JWT can not be found") + return false + } + if !s.isTrustedIssuer(acc.Issuer) { + c.Debugf("Account JWT not signed by trusted operator") + return false + } + if acc.IsExpired() { + c.Debugf("Account JWT has expired") + return false + } + // Verify the signature against the nonce. + if c.opts.Sig == "" { + c.Debugf("Signature missing") + return false + } + sig, err := base64.StdEncoding.DecodeString(c.opts.Sig) + if err != nil { + c.Debugf("Signature not valid base64") + return false + } + pub, err := nkeys.FromPublicKey([]byte(juc.Subject)) + if err != nil { + c.Debugf("User nkey not valid: %v", err) + return false + } + if err := pub.Verify(c.nonce, sig); err != nil { + c.Debugf("Signature not verified") + return false + } + nkey = buildInternalNkeyUser(juc, acc) + c.RegisterNkeyUser(nkey) + + // Check if we need to set an auth timer if the user jwt expires. + c.checkExpiration(juc.Claims()) + return true + } + if nkey != nil { if c.opts.Sig == "" { return false @@ -273,7 +347,7 @@ func (s *Server) isClientAuthorized(c *client) bool { if err != nil { return false } - pub, err := nkeys.FromPublicKey(c.opts.Nkey) + pub, err := nkeys.FromPublicKey([]byte(c.opts.Nkey)) if err != nil { return false } diff --git a/server/client.go b/server/client.go index 94dcddb..fe7fd86 100644 --- a/server/client.go +++ b/server/client.go @@ -26,6 +26,8 @@ import ( "sync" "sync/atomic" "time" + + "github.com/nats-io/jwt" ) // Type of client connection. @@ -128,6 +130,7 @@ const ( DuplicateRoute RouteRemoved ServerShutdown + AuthenticationExpired ) type client struct { @@ -289,6 +292,7 @@ type clientOpts struct { Pedantic bool `json:"pedantic"` TLSRequired bool `json:"tls_required"` Nkey string `json:"nkey,omitempty"` + JWT string `json:"jwt,omitempty"` Sig string `json:"sig,omitempty"` Authorization string `json:"auth_token,omitempty"` Username string `json:"user,omitempty"` @@ -364,16 +368,12 @@ func (c *client) registerWithAccount(acc *Account) error { // If we were previously register, usually to $G, do accounting here to remove. if c.acc != nil { if prev := c.acc.removeClient(c); prev == 1 && c.srv != nil { - c.srv.mu.Lock() - c.srv.activeAccounts-- - c.srv.mu.Unlock() + c.srv.decActiveAccounts() } } // Add in new one. if prev := acc.addClient(c); prev == 0 && c.srv != nil { - c.srv.mu.Lock() - c.srv.activeAccounts++ - c.srv.mu.Unlock() + c.srv.incActiveAccounts() } c.mu.Lock() c.acc = acc @@ -412,14 +412,18 @@ func (c *client) RegisterUser(user *User) { // client with the authenticated user. This is used to map // any permissions into the client and setup accounts. func (c *client) RegisterNkeyUser(user *NkeyUser) { - c.mu.Lock() - defer c.mu.Unlock() - // Register with proper account and sublist. if user.Account != nil { - c.acc = user.Account + if err := c.registerWithAccount(user.Account); err != nil { + c.Errorf("Problem registering with account [%s]", user.Account.Name) + c.sendErr("Failed Account Registration") + return + } } + c.mu.Lock() + defer c.mu.Unlock() + // Assign permissions. if user.Permissions == nil { // Reset perms to nil in case client previously had them. @@ -479,6 +483,20 @@ func (c *client) setPermissions(perms *Permissions) { } } +// Check to see if we have an expiration for the user JWT via base claims. +// FIXME(dlc) - Clear on connect with new JWT. +func (c *client) checkExpiration(claims *jwt.ClaimsData) { + if claims.Expires == 0 { + return + } + tn := time.Now().Unix() + if claims.Expires < tn { + return + } + expiresAt := time.Duration(claims.Expires - tn) + c.setExpirationTimer(expiresAt * time.Second) +} + // This will load up the deny structure used for filtering delivered // messages based on a deny clause for subscriptions. // Lock should be held. @@ -559,7 +577,7 @@ func (c *client) readLoop() { // to process messages, etc. if err := c.parse(b[:n]); err != nil { // handled inline - if err != ErrMaxPayload && err != ErrAuthorization { + if err != ErrMaxPayload && err != ErrAuthentication { c.Errorf("%s", err.Error()) c.closeConnection(ProtocolViolation) } @@ -905,9 +923,9 @@ func (c *client) processConnect(arg []byte) error { } // Check for Auth - if ok := srv.checkAuthorization(c); !ok { + if ok := srv.checkAuthentication(c); !ok { c.authViolation() - return ErrAuthorization + return ErrAuthentication } // Check for Account designation @@ -978,44 +996,68 @@ func (c *client) processConnect(arg []byte) error { return nil } +func (c *client) sendErrAndErr(err string) { + c.sendErr(err) + c.Errorf(err) +} + +func (c *client) sendErrAndDebug(err string) { + c.sendErr(err) + c.Debugf(err) +} + func (c *client) authTimeout() { - c.sendErr(ErrAuthTimeout.Error()) - c.Debugf("Authorization Timeout") + c.sendErrAndDebug("Authentication Timeout") c.closeConnection(AuthenticationTimeout) } +func (c *client) authExpired() { + c.sendErrAndDebug("User Authentication Expired") + c.closeConnection(AuthenticationExpired) +} + +func (c *client) accountAuthExpired() { + c.sendErrAndDebug("Account Authentication Expired") + c.closeConnection(AuthenticationExpired) +} + func (c *client) authViolation() { - var hasNkeys, hasUsers bool + var hasTrustedNkeys, hasNkeys, hasUsers bool if s := c.srv; s != nil { s.mu.Lock() + hasTrustedNkeys = len(s.trustedNkeys) > 0 hasNkeys = s.nkeys != nil hasUsers = s.users != nil s.mu.Unlock() } - if hasNkeys { + if hasTrustedNkeys { + if c.opts.JWT != "" { + c.Errorf("%v", ErrAuthentication) + } else { + c.Errorf("%v", ErrAuthentication) + } + } else if hasNkeys { c.Errorf("%s - Nkey %q", - ErrAuthorization.Error(), + ErrAuthentication.Error(), c.opts.Nkey) } else if hasUsers { c.Errorf("%s - User %q", - ErrAuthorization.Error(), + ErrAuthentication.Error(), c.opts.Username) } else { - c.Errorf(ErrAuthorization.Error()) + c.Errorf(ErrAuthentication.Error()) } c.sendErr("Authorization Violation") c.closeConnection(AuthenticationViolation) } func (c *client) maxConnExceeded() { - c.Errorf(ErrTooManyConnections.Error()) - c.sendErr(ErrTooManyConnections.Error()) + c.sendErrAndErr(ErrTooManyConnections.Error()) c.closeConnection(MaxConnectionsExceeded) } func (c *client) maxSubsExceeded() { - c.Errorf(ErrTooManySubs.Error()) - c.sendErr(ErrTooManySubs.Error()) + c.sendErrAndErr(ErrTooManySubs.Error()) } func (c *client) maxPayloadViolation(sz int, max int64) { @@ -1412,11 +1454,14 @@ func (c *client) addShadowSubscriptions(acc *Account, sub *subscription) error { acc.mu.RLock() // Loop over the import subjects. We have 3 scenarios. If we exact // match or we know the proposed subject is a strict subset of the - // import we can subscribe the subcsription's subject directly. + // import we can subscribe to the subscription's subject directly. // The third scenario is where the proposed subject has a wildcard // and may not be an exact subset, but is a match. Therefore we have to // subscribe to the import subject, not the subscription's subject. for _, im := range acc.imports.streams { + if im.invalid { + continue + } subj := string(sub.subject) if subj == im.prefix+im.from { ims = append(ims, im) @@ -1426,7 +1471,7 @@ func (c *client) addShadowSubscriptions(acc *Account, sub *subscription) error { tokens = tsa[:0] start := 0 for i := 0; i < len(subj); i++ { - //This is not perfect, but the test below will + // This is not perfect, but the test below will // be more exact, this is just to trigger the // additional test. if subj[i] == pwc || subj[i] == fwc { @@ -1551,7 +1596,7 @@ func (c *client) unsubscribe(acc *Account, sub *subscription, force bool) { defer c.mu.Unlock() if !force && sub.max > 0 && sub.nm < sub.max { c.Debugf( - "Deferring actual UNSUB(%s): %d max, %d received\n", + "Deferring actual UNSUB(%s): %d max, %d received", string(sub.subject), sub.max, sub.nm) return } @@ -1704,7 +1749,7 @@ func (c *client) deliverMsg(sub *subscription, mh, msg []byte) bool { // still process the message in hand, otherwise // unsubscribe and drop message on the floor. if sub.nm == sub.max { - client.Debugf("Auto-unsubscribe limit of %d reached for sid '%s'\n", sub.max, string(sub.sid)) + client.Debugf("Auto-unsubscribe limit of %d reached for sid '%s'", sub.max, string(sub.sid)) // Due to defer, reverse the code order so that execution // is consistent with other cases where we unsubscribe. if shouldForward { @@ -1712,7 +1757,7 @@ func (c *client) deliverMsg(sub *subscription, mh, msg []byte) bool { } defer client.unsubscribe(client.acc, sub, true) } else if sub.nm > sub.max { - client.Debugf("Auto-unsubscribe limit [%d] exceeded\n", sub.max) + client.Debugf("Auto-unsubscribe limit [%d] exceeded", sub.max) client.mu.Unlock() client.unsubscribe(client.acc, sub, true) if shouldForward { @@ -1959,7 +2004,7 @@ func (c *client) checkForImportServices(acc *Account, msg []byte) { if c.pa.reply != nil { // We want to remap this to provide anonymity. nrr = c.newServiceReply() - rm.acc.addImplicitServiceImport(acc, string(nrr), string(c.pa.reply), true) + rm.acc.addImplicitServiceImport(acc, string(nrr), string(c.pa.reply), true, nil) } // FIXME(dlc) - Do L1 cache trick from above. rr := rm.acc.sl.Match(rm.to) @@ -1972,7 +2017,7 @@ func (c *client) addSubToRouteTargets(sub *subscription) { c.in.rts = make([]routeTarget, 0, routeTargetInit) } - for i, _ := range c.in.rts { + for i := range c.in.rts { rt := &c.in.rts[i] if rt.sub.client == sub.client { if sub.queue != nil { @@ -2135,7 +2180,7 @@ func (c *client) processMsgResults(acc *Account, r *SublistResult, msg, subject, // We address by index to avoimd struct copy. We have inline structs for memory // layout and cache coherency. - for i, _ := range c.in.rts { + for i := range c.in.rts { rt := &c.in.rts[i] mh := c.msgb[:msgHeadProtoLen] @@ -2237,11 +2282,21 @@ func (c *client) clearAuthTimer() bool { return stopped } -func (c *client) isAuthTimerSet() bool { +// We may reuse atmr for expiring user jwts, +// so check connectReceived. +func (c *client) awaitingAuth() bool { + c.mu.Lock() + authSet := !c.flags.isSet(connectReceived) && c.atmr != nil + c.mu.Unlock() + return authSet +} + +// This will set the atmr for the JWT expiration time. +// We will lock on entry. +func (c *client) setExpirationTimer(d time.Duration) { c.mu.Lock() - isSet := c.atmr != nil + c.atmr = time.AfterFunc(d, c.authExpired) c.mu.Unlock() - return isSet } // Lock should be held diff --git a/server/client_test.go b/server/client_test.go index 7ef980c..bb25a37 100644 --- a/server/client_test.go +++ b/server/client_test.go @@ -717,8 +717,8 @@ func TestAuthorizationTimeout(t *testing.T) { if err != nil { t.Fatalf("Error receiving info from server: %v\n", err) } - if !strings.Contains(l, "Authorization Timeout") { - t.Fatalf("Authorization Timeout response incorrect: %q\n", l) + if !strings.Contains(l, "Authentication Timeout") { + t.Fatalf("Authentication Timeout response incorrect: %q\n", l) } } diff --git a/server/closed_conns_test.go b/server/closed_conns_test.go index 88cdac7..ef11183 100644 --- a/server/closed_conns_test.go +++ b/server/closed_conns_test.go @@ -50,7 +50,7 @@ func TestClosedConnsAccounting(t *testing.T) { s := RunServer(opts) defer s.Shutdown() - wait := 20 * time.Millisecond + wait := time.Second nc, err := nats.Connect(fmt.Sprintf("nats://%s:%d", opts.Host, opts.Port)) if err != nil { diff --git a/server/const.go b/server/const.go index b1144d3..9d4cfbf 100644 --- a/server/const.go +++ b/server/const.go @@ -34,6 +34,9 @@ const ( var ( // gitCommit injected at build gitCommit string + // trustedNkeys is a whitespace separated array of + // trusted operator public nkeys. + trustedNkeys string ) const ( diff --git a/server/errors.go b/server/errors.go index 64cd4bc..5a33394 100644 --- a/server/errors.go +++ b/server/errors.go @@ -22,11 +22,14 @@ var ( // ErrConnectionClosed represents an error condition on a closed connection. ErrConnectionClosed = errors.New("Connection Closed") - // ErrAuthorization represents an error condition on failed authorization. - ErrAuthorization = errors.New("Authorization Error") + // ErrAuthorization represents an error condition on failed authentication. + ErrAuthentication = errors.New("Authentication Error") // ErrAuthTimeout represents an error condition on failed authorization due to timeout. - ErrAuthTimeout = errors.New("Authorization Timeout") + ErrAuthTimeout = errors.New("Authentication Timeout") + + // ErrAuthExpired represents an expired authorization due to timeout. + ErrAuthExpired = errors.New("Authentication Expired") // ErrMaxPayload represents an error condition when the payload is too big. ErrMaxPayload = errors.New("Maximum Payload Exceeded") @@ -65,6 +68,12 @@ var ( // ErrMissingAccount is returned when an account does not exist. ErrMissingAccount = errors.New("Account Missing") + // ErrAccountValidation is returned when an account has failed validation. + ErrAccountValidation = errors.New("Account Validation Failed") + + // ErrNoAccountResolver is returned when we attempt an update but do not have an account resolver. + ErrNoAccountResolver = errors.New("Account Resolver Missing") + // ErrStreamImportAuthorization is returned when a stream import is not authorized. ErrStreamImportAuthorization = errors.New("Stream Import Not Authorized") diff --git a/server/jwt_test.go b/server/jwt_test.go new file mode 100644 index 0000000..e098380 --- /dev/null +++ b/server/jwt_test.go @@ -0,0 +1,1013 @@ +// Copyright 2018 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 +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package server + +import ( + "encoding/base64" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/nats-io/jwt" + "github.com/nats-io/nkeys" +) + +const ( + uJWT = "eyJ0eXAiOiJqd3QiLCJhbGciOiJlZDI1NTE5In0.eyJqdGkiOiJRVzRWWktISEJCUkFaSkFWREg3UjVDSk1RQ1pHWDZJM1FJWEJSMkdWSjRHSVRMRlJRMlpBIiwiaWF0IjoxNTQyMzg1NjMxLCJpc3MiOiJBQ1E1VkpLS1dEM0s1QzdSVkFFMjJNT1hESkFNTEdFTUZJM1NDR1JWUlpKSlFUTU9QTjMzQlhVSyIsIm5hbWUiOiJkZXJlayIsInN1YiI6IlVEMkZMTEdGRVJRVlFRM1NCS09OTkcyUU1JTVRaUUtLTFRVM0FWRzVJM0VRRUZIQlBHUEUyWFFTIiwidHlwZSI6InVzZXIiLCJuYXRzIjp7InB1YiI6e30sInN1YiI6e319fQ.6PmFNn3x0AH3V05oemO28riP63+QTvk9g/Qtt6wBcXJqgW6YSVxk6An1MjvTn1tH7S9tJ0zOIGp7/OLjP1tbBQ" + aJWT = "eyJ0eXAiOiJqd3QiLCJhbGciOiJlZDI1NTE5In0.eyJqdGkiOiJJSTdKSU5JUENVWTZEU1JDSUpZT1daR0k0UlRGNUdCNjVZUUtSNE9RVlBCQlpBNFhCQlhRIiwiaWF0IjoxNTQyMzMxNzgwLCJpc3MiOiJPRDJXMkk0TVZSQTVUR1pMWjJBRzZaSEdWTDNPVEtGV1FKRklYNFROQkVSMjNFNlA0NlMzNDVZWSIsIm5hbWUiOiJmb28iLCJzdWIiOiJBQ1E1VkpLS1dEM0s1QzdSVkFFMjJNT1hESkFNTEdFTUZJM1NDR1JWUlpKSlFUTU9QTjMzQlhVSyIsInR5cGUiOiJhY2NvdW50IiwibmF0cyI6e319.Dg2A1NCJWvXhBQZN9QNHAq1KqsFIKxzLhYvD5yH0DYZPC0gXtdhLkwJ5uiooki6YvzR8UNQZ9XuWgDpNpwryDg" +) + +var ( + uSeed = []byte("SUAIO3FHUX5PNV2LQIIP7TZ3N4L7TX3W53MQGEIVYFIGA635OZCKEYHFLM") + oSeed = []byte("SOAL7GTNI66CTVVNXBNQMG6V2HTDRWC3HGEP7D2OUTWNWSNYZDXWFOX4SU") + aSeed = []byte("SAANRM6JVDEYZTR6DXCWUSDDJHGOHAFITXEQBSEZSY5JENTDVRZ6WNKTTY") +) + +func opTrustBasicSetup() *Server { + kp, _ := nkeys.FromSeed(oSeed) + pub, _ := kp.PublicKey() + opts := defaultServerOptions + opts.TrustedNkeys = []string{string(pub)} + s, _, _, _ := rawSetup(opts) + return s +} + +func buildMemAccResolver(s *Server) { + kp, _ := nkeys.FromSeed(aSeed) + pub, _ := kp.PublicKey() + mr := &memAccResolver{} + mr.Store(string(pub), aJWT) + s.mu.Lock() + s.accResolver = mr + s.mu.Unlock() +} + +func addAccountToMemResolver(s *Server, pub, jwt string) { + s.mu.Lock() + s.accResolver.(*memAccResolver).Store(pub, jwt) + s.mu.Unlock() +} + +func TestJWTUser(t *testing.T) { + s := opTrustBasicSetup() + defer s.Shutdown() + + // Check to make sure we would have an authTimer + if !s.info.AuthRequired { + t.Fatalf("Expect the server to require auth") + } + + c, cr, _ := newClientForServer(s) + + // Don't send jwt field, should fail. + go c.parse([]byte("CONNECT {\"verbose\":true,\"pedantic\":true}\r\nPING\r\n")) + l, _ := cr.ReadString('\n') + if !strings.HasPrefix(l, "-ERR ") { + t.Fatalf("Expected an error") + } + + c, cr, _ = newClientForServer(s) + + // PING needed to flush the +OK/-ERR to us. + // This should fail too since no account resolver is defined. + cs := fmt.Sprintf("CONNECT {\"jwt\":%q,\"sig\":\"%s\",\"verbose\":true,\"pedantic\":true}\r\nPING\r\n", uJWT, "xxx") + go c.parse([]byte(cs)) + l, _ = cr.ReadString('\n') + if !strings.HasPrefix(l, "-ERR ") { + t.Fatalf("Expected an error") + } + + // Ok now let's walk through and make sure all is good. + // We will set the account resolver by hand to a memory resolver. + buildMemAccResolver(s) + + c, cr, l = newClientForServer(s) + + // Sign Nonce + kp, _ := nkeys.FromSeed(uSeed) + + var info nonceInfo + json.Unmarshal([]byte(l[5:]), &info) + sigraw, _ := kp.Sign([]byte(info.Nonce)) + sig := base64.StdEncoding.EncodeToString(sigraw) + + // PING needed to flush the +OK/-ERR to us. + // This should fail too since no account resolver is defined. + cs = fmt.Sprintf("CONNECT {\"jwt\":%q,\"sig\":\"%s\",\"verbose\":true,\"pedantic\":true}\r\nPING\r\n", uJWT, sig) + go c.parse([]byte(cs)) + l, _ = cr.ReadString('\n') + if !strings.HasPrefix(l, "+OK") { + t.Fatalf("Expected an OK, got: %v", l) + } +} + +func TestJWTUserBadTrusted(t *testing.T) { + s := opTrustBasicSetup() + defer s.Shutdown() + + // Check to make sure we would have an authTimer + if !s.info.AuthRequired { + t.Fatalf("Expect the server to require auth") + } + // Now place bad trusted key + s.mu.Lock() + s.trustedNkeys = []string{"bad"} + s.mu.Unlock() + + buildMemAccResolver(s) + + c, cr, l := newClientForServer(s) + + // Sign Nonce + kp, _ := nkeys.FromSeed(uSeed) + + var info nonceInfo + json.Unmarshal([]byte(l[5:]), &info) + sigraw, _ := kp.Sign([]byte(info.Nonce)) + sig := base64.StdEncoding.EncodeToString(sigraw) + + // PING needed to flush the +OK/-ERR to us. + // This should fail too since no account resolver is defined. + cs := fmt.Sprintf("CONNECT {\"jwt\":%q,\"sig\":\"%s\",\"verbose\":true,\"pedantic\":true}\r\nPING\r\n", uJWT, sig) + go c.parse([]byte(cs)) + l, _ = cr.ReadString('\n') + if !strings.HasPrefix(l, "-ERR ") { + t.Fatalf("Expected an error") + } +} + +func TestJWTUserExpired(t *testing.T) { + // Create a new user that we will make sure has expired. + nkp, _ := nkeys.CreateUser() + pub, _ := nkp.PublicKey() + nuc := jwt.NewUserClaims(string(pub)) + nuc.IssuedAt = time.Now().Add(-10 * time.Second).Unix() + nuc.Expires = time.Now().Add(-2 * time.Second).Unix() + + akp, _ := nkeys.FromSeed(aSeed) + jwt, err := nuc.Encode(akp) + if err != nil { + t.Fatalf("Error generating user JWT: %v", err) + } + + s := opTrustBasicSetup() + defer s.Shutdown() + buildMemAccResolver(s) + + c, cr, l := newClientForServer(s) + + // Sign Nonce + var info nonceInfo + json.Unmarshal([]byte(l[5:]), &info) + sigraw, _ := nkp.Sign([]byte(info.Nonce)) + sig := base64.StdEncoding.EncodeToString(sigraw) + + // PING needed to flush the +OK/-ERR to us. + // This should fail too since no account resolver is defined. + cs := fmt.Sprintf("CONNECT {\"jwt\":%q,\"sig\":\"%s\",\"verbose\":true,\"pedantic\":true}\r\nPING\r\n", jwt, sig) + go c.parse([]byte(cs)) + l, _ = cr.ReadString('\n') + if !strings.HasPrefix(l, "-ERR ") { + t.Fatalf("Expected an error") + } +} + +func TestJWTUserExpiresAfterConnect(t *testing.T) { + // Create a new user that we will make sure has expired. + nkp, _ := nkeys.CreateUser() + pub, _ := nkp.PublicKey() + nuc := jwt.NewUserClaims(string(pub)) + nuc.IssuedAt = time.Now().Unix() + nuc.Expires = time.Now().Add(time.Second).Unix() + + akp, _ := nkeys.FromSeed(aSeed) + jwt, err := nuc.Encode(akp) + if err != nil { + t.Fatalf("Error generating user JWT: %v", err) + } + + s := opTrustBasicSetup() + defer s.Shutdown() + buildMemAccResolver(s) + + c, cr, l := newClientForServer(s) + + // Sign Nonce + var info nonceInfo + json.Unmarshal([]byte(l[5:]), &info) + sigraw, _ := nkp.Sign([]byte(info.Nonce)) + sig := base64.StdEncoding.EncodeToString(sigraw) + + // PING needed to flush the +OK/-ERR to us. + // This should fail too since no account resolver is defined. + cs := fmt.Sprintf("CONNECT {\"jwt\":%q,\"sig\":\"%s\",\"verbose\":true,\"pedantic\":true}\r\nPING\r\n", jwt, sig) + go c.parse([]byte(cs)) + l, _ = cr.ReadString('\n') + if !strings.HasPrefix(l, "+OK") { + t.Fatalf("Expected an OK, got: %v", l) + } + l, _ = cr.ReadString('\n') + if !strings.HasPrefix(l, "PONG") { + t.Fatalf("Expected a PONG") + } + + // Now we should expire after 1 second or so. + time.Sleep(time.Second) + + l, _ = cr.ReadString('\n') + if !strings.HasPrefix(l, "-ERR ") { + t.Fatalf("Expected an error") + } + if !strings.Contains(l, "Expired") { + t.Fatalf("Expected 'Expired' to be in the error") + } +} + +func TestJWTUserPermissionClaims(t *testing.T) { + nkp, _ := nkeys.CreateUser() + pub, _ := nkp.PublicKey() + nuc := jwt.NewUserClaims(string(pub)) + + nuc.Permissions.Pub.Allow.Add("foo") + nuc.Permissions.Pub.Allow.Add("bar") + nuc.Permissions.Pub.Deny.Add("baz") + nuc.Permissions.Sub.Allow.Add("foo") + nuc.Permissions.Sub.Allow.Add("bar") + nuc.Permissions.Sub.Deny.Add("baz") + + akp, _ := nkeys.FromSeed(aSeed) + jwt, err := nuc.Encode(akp) + if err != nil { + t.Fatalf("Error generating user JWT: %v", err) + } + + s := opTrustBasicSetup() + defer s.Shutdown() + buildMemAccResolver(s) + + c, cr, l := newClientForServer(s) + + // Sign Nonce + var info nonceInfo + json.Unmarshal([]byte(l[5:]), &info) + sigraw, _ := nkp.Sign([]byte(info.Nonce)) + sig := base64.StdEncoding.EncodeToString(sigraw) + + // PING needed to flush the +OK/-ERR to us. + // This should fail too since no account resolver is defined. + cs := fmt.Sprintf("CONNECT {\"jwt\":%q,\"sig\":\"%s\",\"verbose\":true,\"pedantic\":true}\r\nPING\r\n", jwt, sig) + go c.parse([]byte(cs)) + l, _ = cr.ReadString('\n') + if !strings.HasPrefix(l, "+OK") { + t.Fatalf("Expected an OK, got: %v", l) + } + // Now check client to make sure permissions transferred. + c.mu.Lock() + defer c.mu.Unlock() + + if c.perms == nil { + t.Fatalf("Expected client permissions to be set") + } + + if lpa := c.perms.pub.allow.Count(); lpa != 2 { + t.Fatalf("Expected 2 publish allow subjects, got %d", lpa) + } + if lpd := c.perms.pub.deny.Count(); lpd != 1 { + t.Fatalf("Expected 1 publish deny subjects, got %d", lpd) + } + if lsa := c.perms.sub.allow.Count(); lsa != 2 { + t.Fatalf("Expected 2 subscribe allow subjects, got %d", lsa) + } + if lsd := c.perms.sub.deny.Count(); lsd != 1 { + t.Fatalf("Expected 1 subscribe deny subjects, got %d", lsd) + } +} + +func TestJWTAccountExpired(t *testing.T) { + s := opTrustBasicSetup() + defer s.Shutdown() + buildMemAccResolver(s) + + okp, _ := nkeys.FromSeed(oSeed) + + // Create an account that will be expired. + akp, _ := nkeys.CreateAccount() + apub, _ := akp.PublicKey() + nac := jwt.NewAccountClaims(string(apub)) + nac.IssuedAt = time.Now().Add(-10 * time.Second).Unix() + nac.Expires = time.Now().Add(-2 * time.Second).Unix() + ajwt, err := nac.Encode(okp) + if err != nil { + t.Fatalf("Error generating account JWT: %v", err) + } + + addAccountToMemResolver(s, string(apub), ajwt) + + // Create a new user + nkp, _ := nkeys.CreateUser() + pub, _ := nkp.PublicKey() + nuc := jwt.NewUserClaims(string(pub)) + jwt, err := nuc.Encode(akp) + if err != nil { + t.Fatalf("Error generating user JWT: %v", err) + } + + c, cr, l := newClientForServer(s) + + // Sign Nonce + var info nonceInfo + json.Unmarshal([]byte(l[5:]), &info) + sigraw, _ := nkp.Sign([]byte(info.Nonce)) + sig := base64.StdEncoding.EncodeToString(sigraw) + + // PING needed to flush the +OK/-ERR to us. + // This should fail since the account is expired. + cs := fmt.Sprintf("CONNECT {\"jwt\":%q,\"sig\":\"%s\",\"verbose\":true,\"pedantic\":true}\r\nPING\r\n", jwt, sig) + go c.parse([]byte(cs)) + l, _ = cr.ReadString('\n') + if !strings.HasPrefix(l, "-ERR ") { + t.Fatalf("Expected an error") + } +} + +func TestJWTAccountExpiresAfterConnect(t *testing.T) { + s := opTrustBasicSetup() + defer s.Shutdown() + buildMemAccResolver(s) + + okp, _ := nkeys.FromSeed(oSeed) + + // Create an account that will expire. + akp, _ := nkeys.CreateAccount() + apub, _ := akp.PublicKey() + nac := jwt.NewAccountClaims(string(apub)) + nac.IssuedAt = time.Now().Unix() + nac.Expires = time.Now().Add(time.Second).Unix() + ajwt, err := nac.Encode(okp) + if err != nil { + t.Fatalf("Error generating account JWT: %v", err) + } + + addAccountToMemResolver(s, string(apub), ajwt) + + // Create a new user + nkp, _ := nkeys.CreateUser() + pub, _ := nkp.PublicKey() + nuc := jwt.NewUserClaims(string(pub)) + jwt, err := nuc.Encode(akp) + if err != nil { + t.Fatalf("Error generating user JWT: %v", err) + } + + c, cr, l := newClientForServer(s) + + // Sign Nonce + var info nonceInfo + json.Unmarshal([]byte(l[5:]), &info) + sigraw, _ := nkp.Sign([]byte(info.Nonce)) + sig := base64.StdEncoding.EncodeToString(sigraw) + + // PING needed to flush the +OK/-ERR to us. + cs := fmt.Sprintf("CONNECT {\"jwt\":%q,\"sig\":\"%s\",\"verbose\":true,\"pedantic\":true}\r\nPING\r\n", jwt, sig) + go c.parse([]byte(cs)) + l, _ = cr.ReadString('\n') + if !strings.HasPrefix(l, "+OK") { + t.Fatalf("Expected an OK, got: %v", l) + } + l, _ = cr.ReadString('\n') + if !strings.HasPrefix(l, "PONG") { + t.Fatalf("Expected a PONG") + } + + // Now we should expire after 1 second or so. + time.Sleep(time.Second) + + l, _ = cr.ReadString('\n') + if !strings.HasPrefix(l, "-ERR ") { + t.Fatalf("Expected an error") + } + if !strings.Contains(l, "Expired") { + t.Fatalf("Expected 'Expired' to be in the error") + } + + // Now make sure that accounts that have expired return an error. + c, cr, l = newClientForServer(s) + + // Sign Nonce + json.Unmarshal([]byte(l[5:]), &info) + sigraw, _ = nkp.Sign([]byte(info.Nonce)) + sig = base64.StdEncoding.EncodeToString(sigraw) + + // PING needed to flush the +OK/-ERR to us. + cs = fmt.Sprintf("CONNECT {\"jwt\":%q,\"sig\":\"%s\",\"verbose\":true,\"pedantic\":true}\r\nPING\r\n", jwt, sig) + go c.parse([]byte(cs)) + l, _ = cr.ReadString('\n') + if !strings.HasPrefix(l, "-ERR ") { + t.Fatalf("Expected an error") + } +} + +func TestJWTAccountRenew(t *testing.T) { + s := opTrustBasicSetup() + defer s.Shutdown() + buildMemAccResolver(s) + + okp, _ := nkeys.FromSeed(oSeed) + + // Create an account that has expired. + akp, _ := nkeys.CreateAccount() + apub, _ := akp.PublicKey() + nac := jwt.NewAccountClaims(string(apub)) + nac.IssuedAt = time.Now().Add(-10 * time.Second).Unix() + nac.Expires = time.Now().Add(-2 * time.Second).Unix() + ajwt, err := nac.Encode(okp) + if err != nil { + t.Fatalf("Error generating account JWT: %v", err) + } + + addAccountToMemResolver(s, string(apub), ajwt) + + // Create a new user + nkp, _ := nkeys.CreateUser() + pub, _ := nkp.PublicKey() + nuc := jwt.NewUserClaims(string(pub)) + ujwt, err := nuc.Encode(akp) + if err != nil { + t.Fatalf("Error generating user JWT: %v", err) + } + + c, cr, l := newClientForServer(s) + + // Sign Nonce + var info nonceInfo + json.Unmarshal([]byte(l[5:]), &info) + sigraw, _ := nkp.Sign([]byte(info.Nonce)) + sig := base64.StdEncoding.EncodeToString(sigraw) + + // PING needed to flush the +OK/-ERR to us. + // This should fail since the account is expired. + cs := fmt.Sprintf("CONNECT {\"jwt\":%q,\"sig\":\"%s\",\"verbose\":true,\"pedantic\":true}\r\nPING\r\n", ujwt, sig) + go c.parse([]byte(cs)) + l, _ = cr.ReadString('\n') + if !strings.HasPrefix(l, "-ERR ") { + t.Fatalf("Expected an error") + } + + // Now update with new expiration + nac.IssuedAt = time.Now().Unix() + nac.Expires = time.Now().Add(5 * time.Second).Unix() + ajwt, err = nac.Encode(okp) + if err != nil { + t.Fatalf("Error generating account JWT: %v", err) + } + + // Update the account + addAccountToMemResolver(s, string(apub), ajwt) + acc := s.LookupAccount(string(apub)) + if acc == nil { + t.Fatalf("Expected to retrive the account") + } + s.UpdateAccountClaims(acc, nac) + + // Now make sure we can connect. + c, cr, l = newClientForServer(s) + + // Sign Nonce + json.Unmarshal([]byte(l[5:]), &info) + sigraw, _ = nkp.Sign([]byte(info.Nonce)) + sig = base64.StdEncoding.EncodeToString(sigraw) + + // PING needed to flush the +OK/-ERR to us. + // This should fail too since no account resolver is defined. + cs = fmt.Sprintf("CONNECT {\"jwt\":%q,\"sig\":\"%s\",\"verbose\":true,\"pedantic\":true}\r\nPING\r\n", ujwt, sig) + go c.parse([]byte(cs)) + l, _ = cr.ReadString('\n') + if !strings.HasPrefix(l, "+OK") { + t.Fatalf("Expected an OK, got: %v", l) + } +} + +func TestJWTAccountRenewFromResolver(t *testing.T) { + s := opTrustBasicSetup() + defer s.Shutdown() + buildMemAccResolver(s) + + okp, _ := nkeys.FromSeed(oSeed) + + // Create an account that has expired. + akp, _ := nkeys.CreateAccount() + apub, _ := akp.PublicKey() + nac := jwt.NewAccountClaims(string(apub)) + nac.IssuedAt = time.Now().Add(-10 * time.Second).Unix() + nac.Expires = time.Now().Unix() + ajwt, err := nac.Encode(okp) + if err != nil { + t.Fatalf("Error generating account JWT: %v", err) + } + + addAccountToMemResolver(s, string(apub), ajwt) + // Force it to be loaded by the server and start the expiration timer. + acc := s.LookupAccount(string(apub)) + + // Create a new user + nkp, _ := nkeys.CreateUser() + pub, _ := nkp.PublicKey() + nuc := jwt.NewUserClaims(string(pub)) + ujwt, err := nuc.Encode(akp) + if err != nil { + t.Fatalf("Error generating user JWT: %v", err) + } + + c, cr, l := newClientForServer(s) + + // Sign Nonce + var info nonceInfo + json.Unmarshal([]byte(l[5:]), &info) + sigraw, _ := nkp.Sign([]byte(info.Nonce)) + sig := base64.StdEncoding.EncodeToString(sigraw) + + // PING needed to flush the +OK/-ERR to us. + // This should fail since the account is expired. + cs := fmt.Sprintf("CONNECT {\"jwt\":%q,\"sig\":\"%s\",\"verbose\":true,\"pedantic\":true}\r\nPING\r\n", ujwt, sig) + go c.parse([]byte(cs)) + l, _ = cr.ReadString('\n') + if !strings.HasPrefix(l, "-ERR ") { + t.Fatalf("Expected an error") + } + + // Now update with new expiration + nac.IssuedAt = time.Now().Unix() + nac.Expires = time.Now().Add(5 * time.Second).Unix() + ajwt, err = nac.Encode(okp) + if err != nil { + t.Fatalf("Error generating account JWT: %v", err) + } + + // Update the account + addAccountToMemResolver(s, string(apub), ajwt) + // Make sure the too quick update suppression does not bite us. + acc.updated = time.Now().Add(-1 * time.Hour) + + // Do not update the account directly. The resolver should + // happen automatically. + + // Now make sure we can connect. + c, cr, l = newClientForServer(s) + + // Sign Nonce + json.Unmarshal([]byte(l[5:]), &info) + sigraw, _ = nkp.Sign([]byte(info.Nonce)) + sig = base64.StdEncoding.EncodeToString(sigraw) + + // PING needed to flush the +OK/-ERR to us. + // This should fail too since no account resolver is defined. + cs = fmt.Sprintf("CONNECT {\"jwt\":%q,\"sig\":\"%s\",\"verbose\":true,\"pedantic\":true}\r\nPING\r\n", ujwt, sig) + go c.parse([]byte(cs)) + l, _ = cr.ReadString('\n') + if !strings.HasPrefix(l, "+OK") { + t.Fatalf("Expected an OK, got: %v", l) + } +} + +func TestJWTAccountBasicImportExport(t *testing.T) { + s := opTrustBasicSetup() + defer s.Shutdown() + buildMemAccResolver(s) + + okp, _ := nkeys.FromSeed(oSeed) + + // Create accounts and imports/exports. + fooKP, _ := nkeys.CreateAccount() + fooPub, _ := fooKP.PublicKey() + fooAC := jwt.NewAccountClaims(string(fooPub)) + + // Now create Exports. + streamExport := &jwt.Export{Subject: "foo", Type: jwt.Stream} + streamExport2 := &jwt.Export{Subject: "private", Type: jwt.Stream, TokenReq: true} + serviceExport := &jwt.Export{Subject: "req.echo", Type: jwt.Service, TokenReq: true} + serviceExport2 := &jwt.Export{Subject: "req.add", Type: jwt.Service, TokenReq: true} + + fooAC.Exports.Add(streamExport, streamExport2, serviceExport, serviceExport2) + fooJWT, err := fooAC.Encode(okp) + if err != nil { + t.Fatalf("Error generating account JWT: %v", err) + } + + addAccountToMemResolver(s, string(fooPub), fooJWT) + + acc := s.LookupAccount(string(fooPub)) + if acc == nil { + t.Fatalf("Expected to retrieve the account") + } + + // Check to make sure exports transferred over. + if les := len(acc.exports.streams); les != 2 { + t.Fatalf("Expected exports streams len of 2, got %d", les) + } + if les := len(acc.exports.services); les != 2 { + t.Fatalf("Expected exports services len of 2, got %d", les) + } + _, ok := acc.exports.streams["foo"] + if !ok { + t.Fatalf("Expected to map a stream export") + } + se, ok := acc.exports.services["req.echo"] + if !ok || se == nil { + t.Fatalf("Expected to map a service export") + } + if !se.tokenReq { + t.Fatalf("Expected the service export to require tokens") + } + + barKP, _ := nkeys.CreateAccount() + barPub, _ := barKP.PublicKey() + barAC := jwt.NewAccountClaims(string(barPub)) + + streamImport := &jwt.Import{Account: string(fooPub), Subject: "foo", To: "import.foo", Type: jwt.Stream} + serviceImport := &jwt.Import{Account: string(fooPub), Subject: "req.echo", Type: jwt.Service} + barAC.Imports.Add(streamImport, serviceImport) + barJWT, err := barAC.Encode(okp) + if err != nil { + t.Fatalf("Error generating account JWT: %v", err) + } + addAccountToMemResolver(s, string(barPub), barJWT) + + acc = s.LookupAccount(string(barPub)) + if acc == nil { + t.Fatalf("Expected to retrieve the account") + } + if les := len(acc.imports.streams); les != 1 { + t.Fatalf("Expected imports streams len of 1, got %d", les) + } + // Our service import should have failed without a token. + if les := len(acc.imports.services); les != 0 { + t.Fatalf("Expected imports services len of 0, got %d", les) + } + + // Now add in a bad activation token. + barAC = jwt.NewAccountClaims(string(barPub)) + serviceImport = &jwt.Import{Account: string(fooPub), Subject: "req.echo", Token: "not a token", Type: jwt.Service} + barAC.Imports.Add(serviceImport) + barJWT, err = barAC.Encode(okp) + if err != nil { + t.Fatalf("Error generating account JWT: %v", err) + } + addAccountToMemResolver(s, string(barPub), barJWT) + + s.UpdateAccountClaims(acc, barAC) + + // Our service import should have failed with a bad token. + if les := len(acc.imports.services); les != 0 { + t.Fatalf("Expected imports services len of 0, got %d", les) + } + + // Now make a correct one. + barAC = jwt.NewAccountClaims(string(barPub)) + serviceImport = &jwt.Import{Account: string(fooPub), Subject: "req.echo", Type: jwt.Service} + + activation := jwt.NewActivationClaims(string(barPub)) + activation.Exports = jwt.Exports{} + activation.Exports.Add(&jwt.Export{Subject: "req.echo", Type: jwt.Service}) + actJWT, err := activation.Encode(fooKP) + if err != nil { + t.Fatalf("Error generating activation token: %v", err) + } + serviceImport.Token = actJWT + barAC.Imports.Add(serviceImport) + barJWT, err = barAC.Encode(okp) + if err != nil { + t.Fatalf("Error generating account JWT: %v", err) + } + addAccountToMemResolver(s, string(barPub), barJWT) + s.UpdateAccountClaims(acc, barAC) + // Our service import should have succeeded. + if les := len(acc.imports.services); les != 1 { + t.Fatalf("Expected imports services len of 1, got %d", les) + } + + // Now test url + barAC = jwt.NewAccountClaims(string(barPub)) + serviceImport = &jwt.Import{Account: string(fooPub), Subject: "req.add", Type: jwt.Service} + + activation = jwt.NewActivationClaims(string(barPub)) + activation.Exports = jwt.Exports{} + activation.Exports.Add(&jwt.Export{Subject: "req.add", Type: jwt.Service}) + actJWT, err = activation.Encode(fooKP) + if err != nil { + t.Fatalf("Error generating activation token: %v", err) + } + + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Write([]byte(actJWT)) + })) + defer ts.Close() + + serviceImport.Token = ts.URL + barAC.Imports.Add(serviceImport) + barJWT, err = barAC.Encode(okp) + if err != nil { + t.Fatalf("Error generating account JWT: %v", err) + } + addAccountToMemResolver(s, string(barPub), barJWT) + s.UpdateAccountClaims(acc, barAC) + // Our service import should have succeeded. Should be the only one since we reset. + if les := len(acc.imports.services); les != 1 { + t.Fatalf("Expected imports services len of 1, got %d", les) + } + + // Now streams + barAC = jwt.NewAccountClaims(string(barPub)) + streamImport = &jwt.Import{Account: string(fooPub), Subject: "private", To: "import.private", Type: jwt.Stream} + + barAC.Imports.Add(streamImport) + barJWT, err = barAC.Encode(okp) + if err != nil { + t.Fatalf("Error generating account JWT: %v", err) + } + addAccountToMemResolver(s, string(barPub), barJWT) + s.UpdateAccountClaims(acc, barAC) + // Our stream import should have not succeeded. + if les := len(acc.imports.streams); les != 0 { + t.Fatalf("Expected imports services len of 0, got %d", les) + } + + // Now add in activation. + barAC = jwt.NewAccountClaims(string(barPub)) + streamImport = &jwt.Import{Account: string(fooPub), Subject: "private", To: "import.private", Type: jwt.Stream} + + activation = jwt.NewActivationClaims(string(barPub)) + activation.Exports = jwt.Exports{} + activation.Exports.Add(&jwt.Export{Subject: "private", Type: jwt.Stream}) + actJWT, err = activation.Encode(fooKP) + if err != nil { + t.Fatalf("Error generating activation token: %v", err) + } + streamImport.Token = actJWT + barAC.Imports.Add(streamImport) + barJWT, err = barAC.Encode(okp) + if err != nil { + t.Fatalf("Error generating account JWT: %v", err) + } + addAccountToMemResolver(s, string(barPub), barJWT) + s.UpdateAccountClaims(acc, barAC) + // Our stream import should have not succeeded. + if les := len(acc.imports.streams); les != 1 { + t.Fatalf("Expected imports services len of 1, got %d", les) + } +} + +func TestJWTAccountImportExportUpdates(t *testing.T) { + s := opTrustBasicSetup() + defer s.Shutdown() + buildMemAccResolver(s) + + okp, _ := nkeys.FromSeed(oSeed) + + // Create accounts and imports/exports. + fooKP, _ := nkeys.CreateAccount() + fooPub, _ := fooKP.PublicKey() + fooAC := jwt.NewAccountClaims(string(fooPub)) + streamExport := &jwt.Export{Subject: "foo", Type: jwt.Stream} + + fooAC.Exports.Add(streamExport) + fooJWT, err := fooAC.Encode(okp) + if err != nil { + t.Fatalf("Error generating account JWT: %v", err) + } + addAccountToMemResolver(s, string(fooPub), fooJWT) + + barKP, _ := nkeys.CreateAccount() + barPub, _ := barKP.PublicKey() + barAC := jwt.NewAccountClaims(string(barPub)) + streamImport := &jwt.Import{Account: string(fooPub), Subject: "foo", To: "import", Type: jwt.Stream} + + barAC.Imports.Add(streamImport) + barJWT, err := barAC.Encode(okp) + if err != nil { + t.Fatalf("Error generating account JWT: %v", err) + } + addAccountToMemResolver(s, string(barPub), barJWT) + + // Create a client. + nkp, _ := nkeys.CreateUser() + pub, _ := nkp.PublicKey() + nuc := jwt.NewUserClaims(string(pub)) + ujwt, err := nuc.Encode(barKP) + if err != nil { + t.Fatalf("Error generating user JWT: %v", err) + } + + c, cr, l := newClientForServer(s) + + // Sign Nonce + var info nonceInfo + json.Unmarshal([]byte(l[5:]), &info) + sigraw, _ := nkp.Sign([]byte(info.Nonce)) + sig := base64.StdEncoding.EncodeToString(sigraw) + + // PING needed to flush the +OK/-ERR to us. + // This should fail too since no account resolver is defined. + cs := fmt.Sprintf("CONNECT {\"jwt\":%q,\"sig\":\"%s\",\"verbose\":true,\"pedantic\":true}\r\nSUB import.foo 1\r\nPING\r\n", ujwt, sig) + go c.parse([]byte(cs)) + l, _ = cr.ReadString('\n') + if !strings.HasPrefix(l, "+OK") { + t.Fatalf("Expected an OK, got: %v", l) + } + l, _ = cr.ReadString('\n') + if !strings.HasPrefix(l, "+OK") { + t.Fatalf("Expected an OK, got: %v", l) + } + l, _ = cr.ReadString('\n') + if !strings.HasPrefix(l, "PONG\r\n") { + t.Fatalf("PONG response incorrect: %q\n", l) + } + + checkShadow := func(expected int) { + t.Helper() + c.mu.Lock() + defer c.mu.Unlock() + sub := c.subs["1"] + if ls := len(sub.shadow); ls != expected { + t.Fatalf("Expected shadows to be %d, got %d", expected, ls) + } + } + + // We created a SUB on foo which should create a shadow subscription. + checkShadow(1) + + // Now update bar and remove the import which should make the shadow go away. + barAC = jwt.NewAccountClaims(string(barPub)) + barJWT, err = barAC.Encode(okp) + if err != nil { + t.Fatalf("Error generating account JWT: %v", err) + } + addAccountToMemResolver(s, string(barPub), barJWT) + acc := s.LookupAccount(string(barPub)) + s.UpdateAccountClaims(acc, barAC) + + checkShadow(0) + + // Now add it back and make sure the shadow comes back. + streamImport = &jwt.Import{Account: string(fooPub), Subject: "foo", To: "import", Type: jwt.Stream} + barAC.Imports.Add(streamImport) + barJWT, err = barAC.Encode(okp) + if err != nil { + t.Fatalf("Error generating account JWT: %v", err) + } + addAccountToMemResolver(s, string(barPub), barJWT) + s.UpdateAccountClaims(acc, barAC) + + checkShadow(1) + + // Now change export and make sure it goes away as well. So no exports anymore. + fooAC = jwt.NewAccountClaims(string(fooPub)) + fooJWT, err = fooAC.Encode(okp) + if err != nil { + t.Fatalf("Error generating account JWT: %v", err) + } + addAccountToMemResolver(s, string(fooPub), fooJWT) + s.UpdateAccountClaims(s.LookupAccount(string(fooPub)), fooAC) + + checkShadow(0) + + // Now add it in but with permission required. + streamExport = &jwt.Export{Subject: "foo", Type: jwt.Stream, TokenReq: true} + fooAC.Exports.Add(streamExport) + fooJWT, err = fooAC.Encode(okp) + if err != nil { + t.Fatalf("Error generating account JWT: %v", err) + } + addAccountToMemResolver(s, string(fooPub), fooJWT) + s.UpdateAccountClaims(s.LookupAccount(string(fooPub)), fooAC) + + checkShadow(0) + + // Now put it back as normal. + fooAC = jwt.NewAccountClaims(string(fooPub)) + streamExport = &jwt.Export{Subject: "foo", Type: jwt.Stream} + fooAC.Exports.Add(streamExport) + fooJWT, err = fooAC.Encode(okp) + if err != nil { + t.Fatalf("Error generating account JWT: %v", err) + } + addAccountToMemResolver(s, string(fooPub), fooJWT) + s.UpdateAccountClaims(s.LookupAccount(string(fooPub)), fooAC) + + checkShadow(1) +} + +func TestJWTAccountImportActivationExpires(t *testing.T) { + s := opTrustBasicSetup() + defer s.Shutdown() + buildMemAccResolver(s) + + okp, _ := nkeys.FromSeed(oSeed) + + // Create accounts and imports/exports. + fooKP, _ := nkeys.CreateAccount() + fooPub, _ := fooKP.PublicKey() + fooAC := jwt.NewAccountClaims(string(fooPub)) + streamExport := &jwt.Export{Subject: "foo", Type: jwt.Stream, TokenReq: true} + fooAC.Exports.Add(streamExport) + + fooJWT, err := fooAC.Encode(okp) + if err != nil { + t.Fatalf("Error generating account JWT: %v", err) + } + + addAccountToMemResolver(s, string(fooPub), fooJWT) + + acc := s.LookupAccount(string(fooPub)) + if acc == nil { + t.Fatalf("Expected to retrieve the account") + } + + barKP, _ := nkeys.CreateAccount() + barPub, _ := barKP.PublicKey() + barAC := jwt.NewAccountClaims(string(barPub)) + streamImport := &jwt.Import{Account: string(fooPub), Subject: "foo", To: "import.", Type: jwt.Stream} + + activation := jwt.NewActivationClaims(string(barPub)) + activation.Exports = jwt.Exports{} + activation.Exports.Add(&jwt.Export{Subject: "foo", Type: jwt.Stream}) + activation.IssuedAt = time.Now().Add(-10 * time.Second).Unix() + activation.Expires = time.Now().Add(time.Second).Unix() + actJWT, err := activation.Encode(fooKP) + if err != nil { + t.Fatalf("Error generating activation token: %v", err) + } + streamImport.Token = actJWT + barAC.Imports.Add(streamImport) + barJWT, err := barAC.Encode(okp) + if err != nil { + t.Fatalf("Error generating account JWT: %v", err) + } + addAccountToMemResolver(s, string(barPub), barJWT) + + // Create a client. + nkp, _ := nkeys.CreateUser() + pub, _ := nkp.PublicKey() + nuc := jwt.NewUserClaims(string(pub)) + ujwt, err := nuc.Encode(barKP) + if err != nil { + t.Fatalf("Error generating user JWT: %v", err) + } + + c, cr, l := newClientForServer(s) + + // Sign Nonce + var info nonceInfo + json.Unmarshal([]byte(l[5:]), &info) + sigraw, _ := nkp.Sign([]byte(info.Nonce)) + sig := base64.StdEncoding.EncodeToString(sigraw) + + // PING needed to flush the +OK/-ERR to us. + // This should fail too since no account resolver is defined. + cs := fmt.Sprintf("CONNECT {\"jwt\":%q,\"sig\":\"%s\",\"verbose\":true,\"pedantic\":true}\r\nSUB import.foo 1\r\nPING\r\n", ujwt, sig) + go c.parse([]byte(cs)) + l, _ = cr.ReadString('\n') + if !strings.HasPrefix(l, "+OK") { + t.Fatalf("Expected an OK, got: %v", l) + } + l, _ = cr.ReadString('\n') + if !strings.HasPrefix(l, "+OK") { + t.Fatalf("Expected an OK, got: %v", l) + } + l, _ = cr.ReadString('\n') + if !strings.HasPrefix(l, "PONG\r\n") { + t.Fatalf("PONG response incorrect: %q\n", l) + } + + checkShadow := func(expected int) { + t.Helper() + c.mu.Lock() + defer c.mu.Unlock() + sub := c.subs["1"] + if ls := len(sub.shadow); ls != expected { + t.Fatalf("Expected shadows to be %d, got %d", expected, ls) + } + } + + // We created a SUB on foo which should create a shadow subscription. + checkShadow(1) + + time.Sleep(2 * time.Second) + + // Should have expired and been removed. + checkShadow(0) +} diff --git a/server/monitor.go b/server/monitor.go index 3af97f8..076d986 100644 --- a/server/monitor.go +++ b/server/monitor.go @@ -1039,6 +1039,8 @@ func (reason ClosedState) String() string { return "Route Removed" case ServerShutdown: return "Server Shutdown" + case AuthenticationExpired: + return "Authentication Expired" } return "Unknown State" } diff --git a/server/nkey.go b/server/nkey.go index 943405e..ac7ed82 100644 --- a/server/nkey.go +++ b/server/nkey.go @@ -28,7 +28,8 @@ const ( func (s *Server) nonceRequired() bool { s.optsMu.RLock() defer s.optsMu.RUnlock() - return len(s.opts.Nkeys) > 0 + + return len(s.opts.Nkeys) > 0 || len(s.opts.TrustedNkeys) > 0 } // Generate a nonce for INFO challenge. diff --git a/server/nkey_test.go b/server/nkey_test.go index d3186e6..a08a33f 100644 --- a/server/nkey_test.go +++ b/server/nkey_test.go @@ -36,13 +36,13 @@ type nonceInfo struct { } // This is a seed for a user. We can extract public and private keys from this for testing. -const seed = "SUAKYRHVIOREXV7EUZTBHUHL7NUMHPMAS7QMDU3GTIUWEI5LDNOXD43IZY" +var seed = []byte("SUAKYRHVIOREXV7EUZTBHUHL7NUMHPMAS7QMDU3GTIUWEI5LDNOXD43IZY") func nkeyBasicSetup() (*Server, *client, *bufio.Reader, string) { kp, _ := nkeys.FromSeed(seed) pub, _ := kp.PublicKey() opts := defaultServerOptions - opts.Nkeys = []*NkeyUser{&NkeyUser{Nkey: pub}} + opts.Nkeys = []*NkeyUser{&NkeyUser{Nkey: string(pub)}} return rawSetup(opts) } @@ -50,7 +50,7 @@ func mixedSetup() (*Server, *client, *bufio.Reader, string) { kp, _ := nkeys.FromSeed(seed) pub, _ := kp.PublicKey() opts := defaultServerOptions - opts.Nkeys = []*NkeyUser{&NkeyUser{Nkey: pub}} + opts.Nkeys = []*NkeyUser{&NkeyUser{Nkey: string(pub)}} opts.Users = []*User{&User{Username: "derek", Password: "foo"}} return rawSetup(opts) } diff --git a/server/opts.go b/server/opts.go index c6b0927..be24770 100644 --- a/server/opts.go +++ b/server/opts.go @@ -96,6 +96,7 @@ type Options struct { RQSubsSweep time.Duration `json:"-"` // Deprecated MaxClosedClients int `json:"-"` LameDuckDuration time.Duration `json:"-"` + TrustedNkeys []string `json:"-"` CustomClientAuthentication Authentication `json:"-"` CustomRouterAuthentication Authentication `json:"-"` @@ -418,6 +419,36 @@ func (o *Options) ProcessConfigFile(configFile string) error { continue } o.LameDuckDuration = dur + case "trusted": + switch v.(type) { + case string: + o.TrustedNkeys = []string{v.(string)} + case []string: + o.TrustedNkeys = v.([]string) + case []interface{}: + keys := make([]string, 0, len(v.([]interface{}))) + for _, mv := range v.([]interface{}) { + tk, mv = unwrapValue(mv) + if key, ok := mv.(string); ok { + keys = append(keys, key) + } else { + err := &configErr{tk, fmt.Sprintf("error parsing trusted: unsupported type in array %T", mv)} + errors = append(errors, err) + continue + } + } + o.TrustedNkeys = keys + default: + err := &configErr{tk, fmt.Sprintf("error parsing trusted: unsupported type %T", v)} + errors = append(errors, err) + } + // Do a quick sanity check on keys + for _, key := range o.TrustedNkeys { + if !nkeys.IsValidPublicOperatorKey([]byte(key)) { + err := &configErr{tk, fmt.Sprintf("trust key %q required to be a valid public operator nkey", key)} + errors = append(errors, err) + } + } default: if !tk.IsUsedVariable() { err := &unknownConfigFieldErr{ @@ -688,7 +719,7 @@ func parseAccounts(v interface{}, opts *Options, errors *[]error, warnings *[]er switch strings.ToLower(k) { case "nkey": nk, ok := mv.(string) - if !ok || !nkeys.IsValidPublicAccountKey(nk) { + if !ok || !nkeys.IsValidPublicAccountKey([]byte(nk)) { err := &configErr{tk, fmt.Sprintf("Not a valid public nkey for an account: %q", mv)} *errors = append(*errors, err) continue @@ -1245,7 +1276,7 @@ func parseUsers(mv interface{}, opts *Options, errors *[]error, warnings *[]erro return nil, nil, &configErr{tk, fmt.Sprintf("User entry requires a user and a password")} } else if nkey.Nkey != "" { // Make sure the nkey a proper public nkey for a user.. - if !nkeys.IsValidPublicUserKey(nkey.Nkey) { + if !nkeys.IsValidPublicUserKey([]byte(nkey.Nkey)) { return nil, nil, &configErr{tk, fmt.Sprintf("Not a valid public nkey for a user")} } // If we have user or password defined here that is an error. diff --git a/server/parser.go b/server/parser.go index a06bbae..3ba7e0a 100644 --- a/server/parser.go +++ b/server/parser.go @@ -113,9 +113,9 @@ func (c *client) parse(buf []byte) error { mcl = c.srv.getOpts().MaxControlLine } - // snapshot this, and reset when we receive a + // Snapshot this, and reset when we receive a // proper CONNECT if needed. - authSet := c.isAuthTimerSet() + authSet := c.awaitingAuth() // Move to loop instead of range syntax to allow jumping of i for i = 0; i < len(buf); i++ { @@ -595,7 +595,7 @@ func (c *client) parse(buf []byte) error { } c.drop, c.state = 0, OP_START // Reset notion on authSet - authSet = c.isAuthTimerSet() + authSet = c.awaitingAuth() default: if c.argBuf != nil { c.argBuf = append(c.argBuf, b) @@ -837,7 +837,7 @@ func (c *client) parse(buf []byte) error { authErr: c.authViolation() - return ErrAuthorization + return ErrAuthentication parseErr: c.sendErr("Unknown Protocol Operation") diff --git a/server/reload_test.go b/server/reload_test.go index ddb4c5e..e6281b9 100644 --- a/server/reload_test.go +++ b/server/reload_test.go @@ -2751,8 +2751,8 @@ func TestConfigReloadAccountNKeyUsers(t *testing.T) { synadia := s.LookupAccount("synadia") nats := s.LookupAccount("nats.io") - seed1 := "SUAPM67TC4RHQLKBX55NIQXSMATZDOZK6FNEOSS36CAYA7F7TY66LP4BOM" - seed2 := "SUAIS5JPX4X4GJ7EIIJEQ56DH2GWPYJRPWN5XJEDENJOZHCBLI7SEPUQDE" + seed1 := []byte("SUAPM67TC4RHQLKBX55NIQXSMATZDOZK6FNEOSS36CAYA7F7TY66LP4BOM") + seed2 := []byte("SUAIS5JPX4X4GJ7EIIJEQ56DH2GWPYJRPWN5XJEDENJOZHCBLI7SEPUQDE") kp, _ := nkeys.FromSeed(seed1) pubKey, _ := kp.PublicKey() diff --git a/server/server.go b/server/server.go index 6cfa88b..02fb8e1 100644 --- a/server/server.go +++ b/server/server.go @@ -36,6 +36,8 @@ import ( _ "net/http/pprof" "github.com/nats-io/gnatsd/logger" + "github.com/nats-io/jwt" + "github.com/nats-io/nkeys" ) // Time to wait before starting closing clients when in LD mode. @@ -84,6 +86,7 @@ type Server struct { gacc *Account accounts map[string]*Account activeAccounts int + accResolver AccountResolver clients map[uint64]*client routes map[uint64]*client remotes map[string]*client @@ -142,6 +145,9 @@ type Server struct { // LameDuck mode ldm bool ldmCh chan bool + + // Trusted public operator keys. + trustedNkeys []string } // Make sure all are 64bits for atomic use @@ -186,6 +192,11 @@ func New(opts *Options) *Server { configTime: now, } + // ProcessTrustedNkeys + if !s.processTrustedNkeys() { + return nil + } + s.mu.Lock() defer s.mu.Unlock() @@ -266,6 +277,68 @@ func (s *Server) generateRouteInfoJSON() { s.routeInfoJSON = bytes.Join(pcs, []byte(" ")) } +// isTrustedIssuer will check that the issuer is a trusted public key. +// This is used to make sure and account was signed by a trusted operator. +func (s *Server) isTrustedIssuer(issuer string) bool { + s.mu.Lock() + defer s.mu.Unlock() + for _, tk := range s.trustedNkeys { + if tk == issuer { + return true + } + } + return false +} + +// processTrustedNkeys will process stamped and option based +// trusted nkeys. Returns success. +func (s *Server) processTrustedNkeys() bool { + if trustedNkeys != "" && !s.initStampedTrustedNkeys() { + return false + } else if s.opts.TrustedNkeys != nil { + for _, key := range s.opts.TrustedNkeys { + if !nkeys.IsValidPublicOperatorKey([]byte(key)) { + return false + } + s.trustedNkeys = s.opts.TrustedNkeys + } + } + return true +} + +// checkTrustedNkeyString will check that the string is a valid array +// of public operator nkeys. +func checkTrustedNkeyString(keys string) []string { + tks := strings.Fields(keys) + if len(tks) == 0 { + return nil + } + // Walk all the keys and make sure they are valid. + for _, key := range tks { + if !nkeys.IsValidPublicOperatorKey([]byte(key)) { + return nil + } + } + return tks +} + +// initStampedTrustedNkeys will check the stamped trusted keys +// and will set the server field 'trustedNkeys'. Returns whether +// it succeeded or not. +func (s *Server) initStampedTrustedNkeys() bool { + tks := checkTrustedNkeyString(trustedNkeys) + if len(tks) == 0 { + return false + } + // Check to see if we have an override in options, which will + // cause us to fail also. + if len(s.opts.TrustedNkeys) > 0 { + return false + } + s.trustedNkeys = tks + return true +} + // PrintAndDie is exported for access in other packages. func PrintAndDie(msg string) { fmt.Fprintf(os.Stderr, "%s\n", msg) @@ -329,6 +402,20 @@ func (s *Server) NumActiveAccounts() int { return s.activeAccounts } +// incActiveAccounts() just adds one under lock. +func (s *Server) incActiveAccounts() { + s.mu.Lock() + s.activeAccounts++ + s.mu.Unlock() +} + +// dev=cActiveAccounts() just subtracts one under lock. +func (s *Server) decActiveAccounts() { + s.mu.Lock() + s.activeAccounts-- + s.mu.Unlock() +} + // LookupOrRegisterAccount will return the given account if known or create a new entry. func (s *Server) LookupOrRegisterAccount(name string) (account *Account, isNew bool) { s.mu.Lock() @@ -369,6 +456,9 @@ func (s *Server) registerAccount(acc *Account) { if acc.maxaettl == 0 { acc.maxaettl = DEFAULT_TTL_AE_RESPONSE_MAP } + if acc.clients == nil { + acc.clients = make(map[*client]*client) + } // If we are capable of routing we will track subscription // information for efficient interest propagation. // During config reload, it is possible that account was @@ -387,7 +477,93 @@ func (s *Server) registerAccount(acc *Account) { func (s *Server) LookupAccount(name string) *Account { s.mu.Lock() defer s.mu.Unlock() - return s.accounts[name] + + acc := s.accounts[name] + if acc != nil { + // If we are expired and we have a resolver, then + // return the latest information from the resolver. + if s.accResolver != nil && acc.IsExpired() { + s.UpdateAccount(acc) + } + return acc + } + // If we have a resolver see if it can fetch the account. + return s.FetchAccount(name) +} + +// This will fetch new claims and if found update the account with new claims. +func (s *Server) UpdateAccount(acc *Account) bool { + // TODO(dlc) - Make configurable + if time.Since(acc.updated) < time.Second { + s.Debugf("Requested account update for [%s] ignored, too soon", acc.Name) + return false + } + claimJWT, err := s.fetchRawAccountClaims(acc.Name) + if err != nil { + return false + } + acc.updated = time.Now() + if acc.claimJWT != "" && acc.claimJWT == claimJWT { + s.Debugf("Requested account update for [%s], same claims detected", acc.Name) + return false + } + accClaims, err := s.verifyAccountClaims(claimJWT) + if err == nil && accClaims != nil { + s.UpdateAccountClaims(acc, accClaims) + return true + } + return false +} + +// fetchRawAccountClaims will grab raw account claims iff we have a resolver. +func (s *Server) fetchRawAccountClaims(name string) (string, error) { + accResolver := s.accResolver + if accResolver == nil { + return "", ErrNoAccountResolver + } + // Need to do actual Fetch without the lock. + s.mu.Unlock() + claimJWT, err := accResolver.Fetch(name) + s.mu.Lock() + if err != nil { + return "", err + } + return claimJWT, nil +} + +// fetchAccountClaims will attempt to fetch new claims if a resolver is present. +func (s *Server) fetchAccountClaims(name string) (*jwt.AccountClaims, error) { + claimJWT, err := s.fetchRawAccountClaims(name) + if err != nil { + return nil, err + } + return s.verifyAccountClaims(claimJWT) +} + +// verifyAccountClaims will decode and validate any account claims. +func (s *Server) verifyAccountClaims(claimJWT string) (*jwt.AccountClaims, error) { + if accClaims, err := jwt.DecodeAccountClaims(claimJWT); err != nil { + return nil, err + } else { + vr := jwt.CreateValidationResults() + accClaims.Validate(vr) + if vr.IsBlocking(true) { + return nil, ErrAccountValidation + } + return accClaims, nil + } +} + +// This will fetch an account from a resolver if defined. +// Lock should be held. +func (s *Server) FetchAccount(name string) *Account { + if accClaims, _ := s.fetchAccountClaims(name); accClaims != nil { + if acc := s.buildInternalAccount(accClaims); acc != nil { + s.registerAccount(acc) + return acc + } + } + return nil } // Start up the server, this will block. diff --git a/server/trust_test.go b/server/trust_test.go new file mode 100644 index 0000000..8164aac --- /dev/null +++ b/server/trust_test.go @@ -0,0 +1,123 @@ +// Copyright 2018 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 +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package server + +import ( + "fmt" + "os" + "strings" + "testing" +) + +const ( + t1 = "OBYEOZQ46VZMFMNETBAW2H6VGDSOBLP67VUEZJ5LPR3PIZBWWRIY4UI4" + t2 = "OAHC7NGAHG3YVPTD6QOUFZGPM2OMU6EOS67O2VHBUOA6BJLPTWFHGLKU" +) + +func TestStampedTrustedNkeys(t *testing.T) { + opts := DefaultOptions() + defer func() { trustedNkeys = "" }() + + // Set this to a bad key. We require valid operator public keys. + trustedNkeys = "bad" + if s := New(opts); s != nil { + s.Shutdown() + t.Fatalf("Expected a bad trustedNkeys to return nil server") + } + + trustedNkeys = t1 + s := New(opts) + if s == nil { + t.Fatalf("Expected non-nil server") + } + if len(s.trustedNkeys) != 1 || s.trustedNkeys[0] != t1 { + t.Fatalf("Trusted Nkeys not setup properly") + } + trustedNkeys = strings.Join([]string{t1, t2}, " ") + if s = New(opts); s == nil { + t.Fatalf("Expected non-nil server") + } + if len(s.trustedNkeys) != 2 || s.trustedNkeys[0] != t1 || s.trustedNkeys[1] != t2 { + t.Fatalf("Trusted Nkeys not setup properly") + } + + opts.TrustedNkeys = []string{"OVERRIDE ME"} + if s = New(opts); s != nil { + t.Fatalf("Expected opts.TrustedNkeys to return nil server") + } +} + +func TestTrustedKeysOptions(t *testing.T) { + trustedNkeys = "" + opts := DefaultOptions() + opts.TrustedNkeys = []string{"bad"} + if s := New(opts); s != nil { + s.Shutdown() + t.Fatalf("Expected a bad opts.TrustedNkeys to return nil server") + } + opts.TrustedNkeys = []string{t1} + s := New(opts) + if s == nil { + t.Fatalf("Expected non-nil server") + } + if len(s.trustedNkeys) != 1 || s.trustedNkeys[0] != t1 { + t.Fatalf("Trusted Nkeys not setup properly via options") + } + opts.TrustedNkeys = []string{t1, t2} + if s = New(opts); s == nil { + t.Fatalf("Expected non-nil server") + } + if len(s.trustedNkeys) != 2 || s.trustedNkeys[0] != t1 || s.trustedNkeys[1] != t2 { + t.Fatalf("Trusted Nkeys not setup properly via options") + } +} + +func TestTrustConfigOption(t *testing.T) { + confFileName := createConfFile(t, []byte(fmt.Sprintf("trusted = %q", t1))) + defer os.Remove(confFileName) + opts, err := ProcessConfigFile(confFileName) + if err != nil { + t.Fatalf("Error parsing config: %v", err) + } + if l := len(opts.TrustedNkeys); l != 1 { + t.Fatalf("Expected 1 trusted key, got %d", l) + } + if opts.TrustedNkeys[0] != t1 { + t.Fatalf("Expected trusted key to be %q, got %q", t1, opts.TrustedNkeys[0]) + } + + confFileName = createConfFile(t, []byte(fmt.Sprintf("trusted = [%q, %q]", t1, t2))) + defer os.Remove(confFileName) + opts, err = ProcessConfigFile(confFileName) + if err != nil { + t.Fatalf("Error parsing config: %v", err) + } + if l := len(opts.TrustedNkeys); l != 2 { + t.Fatalf("Expected 2 trusted key, got %d", l) + } + if opts.TrustedNkeys[0] != t1 { + t.Fatalf("Expected trusted key to be %q, got %q", t1, opts.TrustedNkeys[0]) + } + if opts.TrustedNkeys[1] != t2 { + t.Fatalf("Expected trusted key to be %q, got %q", t2, opts.TrustedNkeys[1]) + } + + // Now do a bad one. + confFileName = createConfFile(t, []byte(fmt.Sprintf("trusted = [%q, %q]", t1, "bad"))) + defer os.Remove(confFileName) + _, err = ProcessConfigFile(confFileName) + if err == nil { + t.Fatalf("Expected an error parsing trust keys with a bad key") + } +} diff --git a/server/util.go b/server/util.go index 653b8ec..241e934 100644 --- a/server/util.go +++ b/server/util.go @@ -30,7 +30,7 @@ import ( func genID() string { kp, _ := nkeys.CreateServer() pub, _ := kp.PublicKey() - return pub + return string(pub) } // Ascii numbers 0-9 diff --git a/test/new_routes_test.go b/test/new_routes_test.go index 5914c18..c817fc3 100644 --- a/test/new_routes_test.go +++ b/test/new_routes_test.go @@ -1315,6 +1315,12 @@ func TestNewRouteLargeDistinctQueueSubscribers(t *testing.T) { qsubs[i], _ = ncB.QueueSubscribeSync("foo", qg) } ncB.Flush() + checkFor(t, time.Second, 10*time.Millisecond, func() error { + if ns := srvA.NumSubscriptions(); ns != 100 { + return fmt.Errorf("Number of subscriptions is %d", ns) + } + return nil + }) // Send 10 messages. We should receive 1000 responses. for i := 0; i < 10; i++ { @@ -1322,10 +1328,10 @@ func TestNewRouteLargeDistinctQueueSubscribers(t *testing.T) { } ncA.Flush() - checkFor(t, time.Second, 10*time.Millisecond, func() error { + checkFor(t, 2*time.Second, 10*time.Millisecond, func() error { for i := 0; i < nqsubs; i++ { if n, _, _ := qsubs[i].Pending(); n != 10 { - return fmt.Errorf("Number of messgaes is %d", n) + return fmt.Errorf("Number of messages is %d", n) } } return nil -- 2.19.1