Add proxy_connection_token option to call connect proxy with token

internal/client/handler.go

@@ -339,7 +339,7 @@ func (h *Handler) OnClientConnecting(
 
 	var processClientChannels bool
 
-	if e.Token != "" {
+	if e.Token != "" && !ruleConfig.ProxyConnectionToken {
 		token, err := h.tokenVerifier.VerifyConnectToken(e.Token)
 		if err != nil {
 			if err == jwtverify.ErrTokenExpired {
@@ -483,7 +483,7 @@ func (h *Handler) OnClientConnecting(
 		Credentials:       credentials,
 		Subscriptions:     subscriptions,
 		Data:              data,
-		ClientSideRefresh: !refreshProxyEnabled,
+		ClientSideRefresh: !refreshProxyEnabled || ruleConfig.ProxyConnectionToken,
 	}
 	if newCtx != nil {
 		finalReply.Context = newCtx
@@ -499,6 +499,11 @@ type RefreshExtra struct {
 
 // OnRefresh ...
 func (h *Handler) OnRefresh(c Client, e centrifuge.RefreshEvent, refreshProxyHandler proxy.RefreshHandlerFunc, d proxy.PerCallData) (centrifuge.RefreshReply, RefreshExtra, error) {
+	ruleConfig := h.ruleContainer.Config()
+	if refreshProxyHandler == nil && ruleConfig.ProxyConnectionToken {
+		h.node.Log(centrifuge.NewLogEntry(centrifuge.LogLevelWarn, "no refresh proxy to refresh custom token", map[string]interface{}{"user": c.UserID(), "client": c.ID()}))
+		return centrifuge.RefreshReply{Expired: true}, RefreshExtra{}, nil
+	}
 	if refreshProxyHandler != nil {
 		r, extra, err := refreshProxyHandler(c, e, d)
 		return r, RefreshExtra{Meta: extra.Meta}, err

internal/proxy/connect_handler.go

@@ -57,6 +57,7 @@ func (h *ConnectHandler) Handle(node *centrifuge.Node) ConnectingHandlerFunc {
 			Name:      e.Name,
 			Version:   e.Version,
 			Channels:  e.Channels,
+			Token:     e.Token,
 		}
 		if !h.config.Proxy.UseBase64() {
 			req.Data = e.Data

internal/proxy/refresh_handler.go

@@ -51,7 +51,8 @@ func (h *RefreshHandler) Handle(node *centrifuge.Node) RefreshHandlerFunc {
 			Transport: client.Transport().Name(),
 			Encoding:  getEncoding(h.config.Proxy.UseBase64()),
 
-			User: client.UserID(),
+			User:  client.UserID(),
+			Token: e.Token,
 		}
 		if h.config.Proxy.IncludeMeta() && pcd.Meta != nil {
 			req.Meta = proxyproto.Raw(pcd.Meta)

internal/proxyproto/proxy.proto

@@ -36,6 +36,7 @@ message ConnectRequest {
   string name = 12;
   string version = 13;
   repeated string channels = 14;
+  string token = 15;
 }
 
 message SubscribeOptions {
@@ -80,6 +81,7 @@ message RefreshRequest {
 
   string user = 10;
   bytes meta = 11;
+  string token = 12;
 }
 
 message RefreshResult {

internal/rule/rule.go

@@ -75,6 +75,9 @@ type Config struct {
 	// ClientConnectionRateLimit sets the maximum number of new connections a single Centrifugo
 	// node will accept per second.
 	ClientConnectionRateLimit int
+
+	// ProxyConnectionToken bool
+	ProxyConnectionToken bool
 }
 
 // DefaultConfig has default config options.

main.go

@@ -1414,6 +1414,7 @@ func ruleConfig() rule.Config {
 	cfg.RpcNamespaces = rpcNamespacesFromConfig(v)
 	cfg.ClientConnectionLimit = v.GetInt("client_connection_limit")
 	cfg.ClientConnectionRateLimit = v.GetInt("client_connection_rate_limit")
+	cfg.ProxyConnectionToken = v.GetBool("proxy_connection_token")
 	return cfg
 }