func (r *realm) authClient(sid wamp.ID, client wamp.Peer, details wamp.Dict) (*wamp.Welcome, error) {
	if client.IsLocal() && !r.localAuth {
		authid, _ := wamp.AsString(details["authid"])
		if authid == "" {
			authid = strconv.FormatInt(int64(wamp.GlobalID()), 16)
		}
		details = wamp.Dict{
			"authid":       authid,
			"authrole":     "trusted",
			"authmethod":   "local",
			"authprovider": "static",
			"roles": wamp.Dict{
				wamp.RoleBroker: r.broker.role(),
				wamp.RoleDealer: r.dealer.role(),
			},
		}
		return &wamp.Welcome{Details: details}, nil
	}
	_authmethods, _ := wamp.AsList(details["authmethods"])
	if len(_authmethods) == 0 {
		_authmethods = append(_authmethods, "anonymous")
	}
	var authmethods []string
	for _, val := range _authmethods {
		am, ok := wamp.AsString(val)
		if !ok {
			r.log.Println("!! Could not convert authmethod:", val)
			continue
		}
		if am == "" {
			continue
		}
		authmethods = append(authmethods, am)
	}
	if len(authmethods) == 0 {
		return nil, errors.New("no authentication supplied")
	}
	authr, method := r.getAuthenticator(authmethods)
	if authr == nil {
		return nil, errors.New("could not authenticate with any method")
	}
	welcome, err := authr.Authenticate(sid, details, client)
	if err != nil {
		return nil, err
	}
	welcome.Details["authmethod"] = method
	welcome.Details["roles"] = wamp.Dict{
		wamp.RoleBroker: r.broker.role(),
		wamp.RoleDealer: r.dealer.role(),
	}
	return welcome, nil
}
authClient authenticates the client according to the authmethods in the
HELLO message details and the authenticators available for this realm.
package router
import (
	"errors"
	"fmt"
	"strconv"
	"sync"
	"github.com/gammazero/nexus/v3/router/auth"
	"github.com/gammazero/nexus/v3/stdlog"
	"github.com/gammazero/nexus/v3/transport"
	"github.com/gammazero/nexus/v3/wamp"
)
const metaID = wamp.ID(1)
type testament struct {
	topic   wamp.URI
	args    wamp.List
	kwargs  wamp.Dict
	options wamp.Dict
}
type testamentBucket struct {
	detached  []testament
	destroyed []testament
}
type realm struct {
	broker *broker
	dealer *dealer
	authorizer Authorizer
	authenticators map[string]auth.Authenticator
	clients map[wamp.ID]*wamp.Session
	testaments map[wamp.ID]testamentBucket
	metaPeer  wamp.Peer
	metaSess  *wamp.Session
	metaIDGen *wamp.IDGen
	actionChan chan func()
	waitHandlers sync.WaitGroup
	metaProcMap map[wamp.ID]func(*wamp.Invocation) wamp.Message
	metaDone    chan struct{}
	closed    bool
	closeLock sync.Mutex
	log   stdlog.StdLog
	debug bool
	localAuth  bool
	localAuthz bool
	metaStrict     bool
	metaIncDetails []string
	enableMetaKill   bool
	enableMetaModify bool
}
var (
	shutdownGoodbye = &wamp.Goodbye{
		Reason:  wamp.ErrSystemShutdown,
		Details: wamp.Dict{},
	}
)
func newRealm(config *RealmConfig, broker *broker, dealer *dealer, logger stdlog.StdLog, debug bool) (*realm, error) {
	if !config.URI.ValidURI(config.StrictURI, "") {
		return nil, fmt.Errorf(
			"invalid realm URI %v (URI strict checking %v)", config.URI, config.StrictURI)
	}
	r := &realm{
		broker:      broker,
		dealer:      dealer,
		authorizer:  config.Authorizer,
		clients:     map[wamp.ID]*wamp.Session{},
		testaments:  map[wamp.ID]testamentBucket{},
		actionChan:  make(chan func()),
		metaIDGen:   new(wamp.IDGen),
		metaDone:    make(chan struct{}),
		metaProcMap: make(map[wamp.ID]func(*wamp.Invocation) wamp.Message, 9),
		log:         logger,
		debug:       debug,
		localAuth:   config.RequireLocalAuth,
		localAuthz:  config.RequireLocalAuthz,
		metaStrict:  config.MetaStrict,
		enableMetaKill:   config.EnableMetaKill,
		enableMetaModify: config.EnableMetaModify,
	}
	if debug {
		if r.enableMetaKill {
			r.log.Println("Session meta kill procedures enabled")
		}
		if r.enableMetaKill {
			r.log.Println("Session meta modify_details procedure enabled")
		}
	}
	if r.metaStrict && len(config.MetaIncludeSessionDetails) != 0 {
		r.metaIncDetails = make([]string, len(config.MetaIncludeSessionDetails))
		copy(r.metaIncDetails, config.MetaIncludeSessionDetails)
	}
	r.authenticators = map[string]auth.Authenticator{}
	for _, auth := range config.Authenticators {
		r.authenticators[auth.AuthMethod()] = auth
	}
	if config.AnonymousAuth {
		if _, ok := r.authenticators["anonymous"]; !ok {
			r.authenticators["anonymous"] = &auth.AnonymousAuth{
				AuthRole: "anonymous",
			}
		}
	}
	return r, nil
}
func (r *realm) waitReady() {
	sync := make(chan struct{})
	r.actionChan <- func() {
		close(sync)
	}
	<-sync
}
func (r *realm) close() {
	r.closeLock.Lock()
	defer r.closeLock.Unlock()
	if r.closed {
		return
	}
	r.closed = true
	r.waitReady()
	sync := make(chan struct{})
	r.actionChan <- func() {
		for _, c := range r.clients {
			c.EndRecv(shutdownGoodbye)
		}
		close(sync)
	}
	<-sync
	r.waitHandlers.Wait()
	r.metaSess.EndRecv(shutdownGoodbye)
	<-r.metaDone
	r.dealer.close()
	r.broker.close()
	close(r.actionChan)
}
func (r *realm) run() {
	r.createMetaSession()
	r.registerMetaProcedure(wamp.MetaProcSessionCount, r.sessionCount)
	r.registerMetaProcedure(wamp.MetaProcSessionList, r.sessionList)
	r.registerMetaProcedure(wamp.MetaProcSessionGet, r.sessionGet)
	if r.enableMetaKill {
		r.registerMetaProcedure(wamp.MetaProcSessionKill, r.sessionKill)
		r.registerMetaProcedure(wamp.MetaProcSessionKillByAuthid, r.sessionKillByAuthid)
		r.registerMetaProcedure(wamp.MetaProcSessionKillByAuthrole, r.sessionKillByAuthrole)
		r.registerMetaProcedure(wamp.MetaProcSessionKillAll, r.sessionKillAll)
	}
	if r.enableMetaModify {
		r.registerMetaProcedure(wamp.MetaProcSessionModifyDetails, r.sessionModifyDetails)
	}
	r.registerMetaProcedure(wamp.MetaProcRegList, r.dealer.regList)
	r.registerMetaProcedure(wamp.MetaProcRegLookup, r.dealer.regLookup)
	r.registerMetaProcedure(wamp.MetaProcRegMatch, r.dealer.regMatch)
	r.registerMetaProcedure(wamp.MetaProcRegGet, r.dealer.regGet)
	r.registerMetaProcedure(wamp.MetaProcRegListCallees, r.dealer.regListCallees)
	r.registerMetaProcedure(wamp.MetaProcRegCountCallees, r.dealer.regCountCallees)
	r.registerMetaProcedure(wamp.MetaProcSubList, r.broker.subList)
	r.registerMetaProcedure(wamp.MetaProcSubLookup, r.broker.subLookup)
	r.registerMetaProcedure(wamp.MetaProcSubMatch, r.broker.subMatch)
	r.registerMetaProcedure(wamp.MetaProcSubGet, r.broker.subGet)
	r.registerMetaProcedure(wamp.MetaProcSubListSubscribers, r.broker.subListSubscribers)
	r.registerMetaProcedure(wamp.MetaProcSubCountSubscribers, r.broker.subCountSubscribers)
	r.registerMetaProcedure(wamp.MetaProcSessionAddTestament, r.testamentAdd)
	r.registerMetaProcedure(wamp.MetaProcSessionFlushTestaments, r.testamentFlush)
	go r.metaProcedureHandler()
	for action := range r.actionChan {
		action()
	}
}
func (r *realm) createMetaSession() {
	cli, rtr := transport.LinkedPeers()
	r.metaPeer = cli
	r.dealer.setMetaPeer(cli)
	r.metaSess = wamp.NewSession(rtr, metaID, wamp.Dict{"authrole": "trusted"}, nil)
	go r.handleInboundMessages(r.metaSess)
	if r.debug {
		r.log.Println("Started meta-session", r.metaSess)
	}
}
func (r *realm) onJoin(sess *wamp.Session) {
	r.waitHandlers.Add(1)
	sync := make(chan struct{})
	r.actionChan <- func() {
		r.clients[sess.ID] = sess
		close(sync)
	}
	<-sync
	sess.Lock()
	output := r.cleanSessionDetails(sess.Details)
	sess.Unlock()
	r.metaPeer.Send(&wamp.Publish{
		Request:   wamp.GlobalID(),
		Topic:     wamp.MetaEventSessionOnJoin,
		Arguments: wamp.List{output},
	})
}
func (r *realm) onLeave(sess *wamp.Session, shutdown, killAll bool) {
	var testaments testamentBucket
	var hasTstm bool
	sync := make(chan struct{})
	r.actionChan <- func() {
		delete(r.clients, sess.ID)
		testaments, hasTstm = r.testaments[sess.ID]
		if hasTstm {
			delete(r.testaments, sess.ID)
		}
		if !shutdown {
			r.dealer.removeSession(sess)
			r.broker.removeSession(sess)
		}
		close(sync)
	}
	<-sync
	defer r.waitHandlers.Done()
	if shutdown || killAll {
		return
	}
	if hasTstm {
		sendTestaments := func(testaments []testament) {
			for i := range testaments {
				r.metaPeer.Send(&wamp.Publish{
					Request:     wamp.GlobalID(),
					Topic:       testaments[i].topic,
					Arguments:   testaments[i].args,
					ArgumentsKw: testaments[i].kwargs,
					Options:     testaments[i].options,
				})
			}
		}
		sendTestaments(testaments.detached)
		sendTestaments(testaments.destroyed)
	}
	r.metaPeer.Send(&wamp.Publish{
		Request: wamp.GlobalID(),
		Topic:   wamp.MetaEventSessionOnLeave,
		Arguments: wamp.List{
			sess.ID,
			sess.Details["authid"],
			sess.Details["authrole"]},
	})
}
func (r *realm) handleSession(sess *wamp.Session) error {
	r.closeLock.Lock()
	if r.closed {
		r.closeLock.Unlock()
		err := errors.New("realm closed")
		return err
	}
	r.onJoin(sess)
	r.closeLock.Unlock()
	if r.debug {
		r.log.Println("Handling messages for session", sess)
	}
	go func() {
		shutdown, killAll, err := r.handleInboundMessages(sess)
		if err != nil {
			abortMsg := wamp.Abort{
				Reason:  wamp.ErrProtocolViolation,
				Details: wamp.Dict{"error": err.Error()},
			}
			r.log.Println("Aborting session", sess, ":", err)
			sess.TrySend(&abortMsg)
		}
		r.onLeave(sess, shutdown, killAll)
		sess.Close()
	}()
	return nil
}
func (r *realm) handleInboundMessages(sess *wamp.Session) (bool, bool, error) {
	if r.debug {
		defer r.log.Println("Ended session", sess)
	}
	recv := sess.Recv()
	recvDone := sess.RecvDone()
	for {
		var msg wamp.Message
		var open bool
		select {
		case msg, open = <-recv:
			if !open {
				r.log.Println("Lost", sess)
				return false, false, nil
			}
		case <-recvDone:
			goodbye := sess.Goodbye()
			switch goodbye {
			case shutdownGoodbye, wamp.NoGoodbye:
				if r.debug {
					r.log.Printf("Stop session %s: system shutdown", sess)
				}
				sess.TrySend(goodbye)
				return true, false, nil
			}
			if r.debug {
				r.log.Printf("Kill session %s: %s", sess, goodbye.Reason)
			}
			var killAll bool
			if _, ok := goodbye.Details["all"]; ok {
				killAll = true
			}
			sess.TrySend(goodbye)
			return false, killAll, nil
		}
		if r.debug {
			r.log.Printf("Session %s submitting %s: %+v", sess,
				msg.MessageType(), msg)
		}
		if r.authorizer != nil && sess != r.metaSess && !r.authzMessage(sess, msg) {
			continue
		}
		switch msg := msg.(type) {
		case *wamp.Publish:
			r.broker.publish(sess, msg)
		case *wamp.Subscribe:
			r.broker.subscribe(sess, msg)
		case *wamp.Unsubscribe:
			r.broker.unsubscribe(sess, msg)
		case *wamp.Register:
			r.dealer.register(sess, msg)
		case *wamp.Unregister:
			r.dealer.unregister(sess, msg)
		case *wamp.Call:
			r.dealer.call(sess, msg)
		case *wamp.Yield:
			r.dealer.yield(sess, msg)
		case *wamp.Cancel:
			r.dealer.cancel(sess, msg)
		case *wamp.Error:
			if msg.Type != wamp.INVOCATION {
				return false, false, fmt.Errorf("invalid ERROR received: %v", msg)
			}
			r.dealer.error(msg)
		case *wamp.Goodbye:
			sess.TrySend(&wamp.Goodbye{
				Reason:  wamp.ErrGoodbyeAndOut,
				Details: wamp.Dict{},
			})
			if r.debug {
				r.log.Println("GOODBYE from session", sess, "reason:",
					msg.Reason)
			}
			return false, false, nil
		default:
			return false, false, fmt.Errorf("unexpected %v", msg.MessageType())
		}
	}
}
func (r *realm) authzMessage(sess *wamp.Session, msg wamp.Message) bool {
	if sess.Peer.IsLocal() && !r.localAuthz {
		return true
	}
	safeSession := &wamp.Session{
		ID:      sess.ID,
		Details: sess.Details,
	}
	sess.Lock()
	isAuthz, err := r.authorizer.Authorize(safeSession, msg)
	sess.Unlock()
	if !isAuthz {
		skipResponse := false
		errRsp := &wamp.Error{Type: msg.MessageType(), Details: wamp.Dict{}}
		switch msg := msg.(type) {
		case *wamp.Publish:
			if pubAck, _ := msg.Options[wamp.OptAcknowledge].(bool); !pubAck {
				skipResponse = true
			}
			errRsp.Request = msg.Request
		case *wamp.Subscribe:
			errRsp.Request = msg.Request
		case *wamp.Unsubscribe:
			errRsp.Request = msg.Request
		case *wamp.Register:
			errRsp.Request = msg.Request
		case *wamp.Unregister:
			errRsp.Request = msg.Request
		case *wamp.Call:
			errRsp.Request = msg.Request
		case *wamp.Cancel:
			errRsp.Request = msg.Request
		case *wamp.Yield:
			errRsp.Request = msg.Request
		}
		if err != nil {
			errRsp.Error = wamp.ErrAuthorizationFailed
			errRsp.Arguments = wamp.List{err.Error()}
			r.log.Println("Client", sess, "authorization failed:", err)
		} else {
			errRsp.Error = wamp.ErrNotAuthorized
			r.log.Println("Client", sess, msg.MessageType(), "not authorized")
		}
		if !skipResponse {
			err = sess.TrySend(errRsp)
			if err != nil {
				r.log.Println("!!! client blocked, could not send authz error")
			}
		}
		return false
	}
	return true
}