package jwtverify import ( "context" "crypto/ecdsa" "crypto/rsa" "encoding/base64" "encoding/json" "errors" "fmt" "regexp" "strings" "sync" "time" "github.com/centrifugal/centrifugo/v6/internal/config" "github.com/centrifugal/centrifugo/v6/internal/jwks" "github.com/centrifugal/centrifugo/v6/internal/subsource" "github.com/centrifugal/centrifugo/v6/internal/tools" "github.com/centrifugal/centrifuge" "github.com/cristalhq/jwt/v5" "github.com/rakutentech/jwk-go/okp" "github.com/rs/zerolog/log" "github.com/tidwall/gjson" ) type VerifierConfig struct { // HMACSecretKey is a secret key used to validate connection and subscription // tokens generated using HMAC. Zero value means that HMAC tokens won't be allowed. HMACSecretKey string // HMACPreviousSecretKey is the previous HMAC secret key used during key rotation. // When set, tokens that fail verification with HMACSecretKey will be retried with // this key, enabling seamless key rotation without mass disconnects. HMACPreviousSecretKey string // HMACPreviousSecretKeyValidUntil is an optional Unix timestamp. If set, the previous // HMAC key will only be used for verification if the current time is before this timestamp. HMACPreviousSecretKeyValidUntil int64 // RSAPublicKey is a public key used to validate connection and subscription // tokens generated using RSA. Zero value means that RSA tokens won't be allowed. RSAPublicKey *rsa.PublicKey // ECDSAPublicKey is a public key used to validate connection and subscription // tokens generated using ECDSA. Zero value means that ECDSA tokens won't be allowed. ECDSAPublicKey *ecdsa.PublicKey // JWKSPublicEndpoint is a public url used to validate connection and subscription // tokens generated using rotating RSA public keys. Zero value means that JSON Web Key Sets // extension won't be used. JWKSPublicEndpoint string // Audience when set will enable audience token check. See // https://datatracker.ietf.org/doc/html/rfc7519#section-4.1.3. Audience string // AudienceRegex allows setting Audience in form of Go language regex pattern. Regex groups // may be then used in constructing JWKSPublicEndpoint. AudienceRegex string // Issuer when set will enable a check that token issuer matches configured string. // See https://datatracker.ietf.org/doc/html/rfc7519#section-4.1.1. Issuer string // IssuerRegex allows setting Issuer in form of Go language regex pattern. Regex groups // may be then used in constructing JWKSPublicEndpoint. IssuerRegex string // UserIDClaim allows overriding default claim used to extract user ID from token. // By default, Centrifugo uses "sub" and we recommend keeping the default if possible. UserIDClaim string // InsecureSkipJWKSEndpointSafetyCheck disables the config-time safety validation of // JWKS endpoint URL templates. This is INSECURE and exists only as a temporary escape // hatch for users upgrading with existing permissive regex patterns. This option will // be removed in a future release. InsecureSkipJWKSEndpointSafetyCheck bool } func (c VerifierConfig) Validate() error { if c.Audience != "" && c.AudienceRegex != "" { return errors.New("can not use both token_audience and token_audience_regex, configure only one of them") } if c.Issuer != "" && c.IssuerRegex != "" { return errors.New("can not use both token_issuer and token_issuer_regex, configure only one of them") } if c.JWKSPublicEndpoint != "" { if err := validateJWKSEndpointSafety(c.JWKSPublicEndpoint, c.IssuerRegex, c.AudienceRegex); err != nil { if c.InsecureSkipJWKSEndpointSafetyCheck { log.Warn().Err(err).Msg("JWKS endpoint template safety check skipped — this is INSECURE and the escape hatch will be removed in a future release, please update your regex to use an explicit list of allowed values") } else { return err } } } return nil } func NewTokenVerifierJWT(config VerifierConfig, cfgContainer *config.Container) (*VerifierJWT, error) { if err := config.Validate(); err != nil { return nil, fmt.Errorf("error validating token verifier config: %w", err) } var audienceRe *regexp.Regexp var issuerRe *regexp.Regexp var err error if config.AudienceRegex != "" { audienceRe, err = regexp.Compile(config.AudienceRegex) if err != nil { return nil, fmt.Errorf("error compiling audience regex: %w", err) } } if config.IssuerRegex != "" { issuerRe, err = regexp.Compile(config.IssuerRegex) if err != nil { return nil, fmt.Errorf("error compiling issuer regex: %w", err) } } verifier := &VerifierJWT{ cfgContainer: cfgContainer, issuer: config.Issuer, issuerRe: issuerRe, audience: config.Audience, audienceRe: audienceRe, userIDClaim: config.UserIDClaim, } if config.JWKSPublicEndpoint != "" { mng, err := jwks.NewManager(config.JWKSPublicEndpoint) if err != nil { return nil, fmt.Errorf("error creating JWK manager: %w", err) } verifier.jwksManager = &jwksManager{mng} log.Info().Str("endpoint", strings.Join(tools.RedactedLogURLs(config.JWKSPublicEndpoint), ",")). Msg("JWKS manager created") } else { alg, err := newAlgorithms(config.HMACSecretKey, config.RSAPublicKey, config.ECDSAPublicKey) if err != nil { return nil, fmt.Errorf("error initializing token algorithms: %w", err) } verifier.algorithms = alg if config.HMACPreviousSecretKey != "" { prevAlg, err := newAlgorithms(config.HMACPreviousSecretKey, nil, nil) if err != nil { return nil, fmt.Errorf("error initializing previous HMAC token algorithms: %w", err) } verifier.previousHMACAlgorithms = prevAlg verifier.hmacPreviousSecretKeyValidUntil = config.HMACPreviousSecretKeyValidUntil } } return verifier, nil } type VerifierJWT struct { mu sync.RWMutex jwksManager *jwksManager algorithms *algorithms previousHMACAlgorithms *algorithms hmacPreviousSecretKeyValidUntil int64 cfgContainer *config.Container audience string audienceRe *regexp.Regexp issuer string issuerRe *regexp.Regexp userIDClaim string } var ( ErrTokenExpired = errors.New("token expired") ErrInvalidToken = errors.New("invalid token") errPublicKeyInvalid = errors.New("public key is invalid") errUnsupportedAlgorithm = errors.New("unsupported JWT algorithm") errDisabledAlgorithm = errors.New("disabled JWT algorithm") ) // BoolValue allows override boolean option. type BoolValue struct { Value bool `json:"value,omitempty"` } // SubscribeOptionOverride to override configured behaviour of subscriptions. type SubscribeOptionOverride struct { // Presence turns on participating in channel presence. Presence *BoolValue `json:"presence,omitempty"` // JoinLeave enables sending Join and Leave messages for this client in channel. JoinLeave *BoolValue `json:"join_leave,omitempty"` // ForcePushJoinLeave forces sending join/leave for this client. ForcePushJoinLeave *BoolValue `json:"force_push_join_leave,omitempty"` // ForcePositioning on says that client will additionally sync its position inside // a stream to prevent message loss. Make sure you are enabling ForcePositioning in channels // that maintain Publication history stream. When ForcePositioning is on Centrifuge will // include StreamPosition information to subscribe response - for a client to be able // to manually track its position inside a stream. ForcePositioning *BoolValue `json:"force_positioning,omitempty"` // ForceRecovery turns on recovery option for a channel. In this case client will try to // recover missed messages automatically upon resubscribe to a channel after reconnect // to a server. This option also enables client position tracking inside a stream // (like ForcePositioning option) to prevent occasional message loss. Make sure you are using // ForceRecovery in channels that maintain Publication history stream. ForceRecovery *BoolValue `json:"force_recovery,omitempty"` } // SubscribeOptions define per-subscription options. type SubscribeOptions struct { // Info defines custom channel information, zero value means no channel information. Info json.RawMessage `json:"info,omitempty"` // Base64Info is like Info but for binary. Base64Info string `json:"b64info,omitempty"` // Data to send to a client with Subscribe Push. Data json.RawMessage `json:"data,omitempty"` // Base64Data is like Data but for binary data. Base64Data string `json:"b64data,omitempty"` // Override channel options can contain channel options overrides. Override *SubscribeOptionOverride `json:"override,omitempty"` } type ConnectTokenClaims struct { ExpireAt *int64 `json:"expire_at,omitempty"` Info json.RawMessage `json:"info,omitempty"` Base64Info string `json:"b64info,omitempty"` Channels []string `json:"channels,omitempty"` Subs map[string]SubscribeOptions `json:"subs,omitempty"` Meta json.RawMessage `json:"meta,omitempty"` // Channel must never be set in connection tokens. We check this on verifying. Channel string `json:"channel,omitempty"` jwt.RegisteredClaims } type SubscribeTokenClaims struct { jwt.RegisteredClaims SubscribeOptions Channel string `json:"channel,omitempty"` Client string `json:"client,omitempty"` ExpireAt *int64 `json:"expire_at,omitempty"` } type jwksManager struct{ *jwks.Manager } func (j *jwksManager) verify(token *jwt.Token, tokenVars map[string]any) error { kid := token.Header().KeyID key, err := j.Manager.FetchKey(context.Background(), kid, tokenVars) if err != nil { return err } if key.Kty != "RSA" && key.Kty != "EC" && key.Kty != "OKP" { return errUnsupportedAlgorithm } spec, err := key.ParseKeySpec() if err != nil { return fmt.Errorf("error parsing key spec: %w", err) } switch key.Kty { case "RSA": pubKey, ok := spec.Key.(*rsa.PublicKey) if !ok { return errPublicKeyInvalid } keySetAlgorithm := spec.Algorithm if keySetAlgorithm == "" { // "alg" is optional: https://datatracker.ietf.org/doc/html/rfc7517#section-4.4. switch token.Header().Algorithm { case jwt.RS256, jwt.RS384, jwt.RS512: // Only allow algorithms from RSA family. keySetAlgorithm = string(token.Header().Algorithm) default: return fmt.Errorf("%w: no match in algorithms", errUnsupportedAlgorithm) } } verifier, err := jwt.NewVerifierRS(jwt.Algorithm(keySetAlgorithm), pubKey) if err != nil { return fmt.Errorf("%w: %s", errUnsupportedAlgorithm, keySetAlgorithm) } return verifier.Verify(token) case "EC": pubKey, ok := spec.Key.(*ecdsa.PublicKey) if !ok { return errPublicKeyInvalid } keySetAlgorithm := spec.Algorithm if keySetAlgorithm == "" { // "alg" is optional: https://datatracker.ietf.org/doc/html/rfc7517#section-4.4. switch token.Header().Algorithm { case jwt.ES256, jwt.ES384, jwt.ES512: // Only allow algorithms from EC family. keySetAlgorithm = string(token.Header().Algorithm) default: return fmt.Errorf("%w: no match in algorithms", errUnsupportedAlgorithm) } } verifier, err := jwt.NewVerifierES(jwt.Algorithm(keySetAlgorithm), pubKey) if err != nil { return fmt.Errorf("%w: %s", errUnsupportedAlgorithm, keySetAlgorithm) } return verifier.Verify(token) case "OKP": pubKey, ok := spec.Key.(okp.Ed25519) if !ok { return errPublicKeyInvalid } verifier, err := jwt.NewVerifierEdDSA(pubKey.PublicKey()) if err != nil { return errUnsupportedAlgorithm } return verifier.Verify(token) default: return errUnsupportedAlgorithm } } type algorithms struct { HS256 jwt.Verifier HS384 jwt.Verifier HS512 jwt.Verifier RS256 jwt.Verifier RS384 jwt.Verifier RS512 jwt.Verifier ES256 jwt.Verifier ES384 jwt.Verifier ES512 jwt.Verifier } func newAlgorithms(tokenHMACSecretKey string, rsaPubKey *rsa.PublicKey, ecdsaPubKey *ecdsa.PublicKey) (*algorithms, error) { alg := &algorithms{} var algorithms []string // HMAC SHA. if tokenHMACSecretKey != "" { verifierHS256, err := jwt.NewVerifierHS(jwt.HS256, []byte(tokenHMACSecretKey)) if err != nil { return nil, err } verifierHS384, err := jwt.NewVerifierHS(jwt.HS384, []byte(tokenHMACSecretKey)) if err != nil { return nil, err } verifierHS512, err := jwt.NewVerifierHS(jwt.HS512, []byte(tokenHMACSecretKey)) if err != nil { return nil, err } alg.HS256 = verifierHS256 alg.HS384 = verifierHS384 alg.HS512 = verifierHS512 algorithms = append(algorithms, []string{"HS256", "HS384", "HS512"}...) } // RSA. if rsaPubKey != nil { if verifierRS256, err := jwt.NewVerifierRS(jwt.RS256, rsaPubKey); err != nil { if !errors.Is(err, jwt.ErrInvalidKey) { return nil, err } } else { alg.RS256 = verifierRS256 algorithms = append(algorithms, "RS256") } if verifierRS384, err := jwt.NewVerifierRS(jwt.RS384, rsaPubKey); err != nil { if !errors.Is(err, jwt.ErrInvalidKey) { return nil, err } } else { alg.RS384 = verifierRS384 algorithms = append(algorithms, "RS384") } if verifierRS512, err := jwt.NewVerifierRS(jwt.RS512, rsaPubKey); err != nil { if !errors.Is(err, jwt.ErrInvalidKey) { return nil, err } } else { alg.RS512 = verifierRS512 algorithms = append(algorithms, "RS512") } } // ECDSA. if ecdsaPubKey != nil { if verifierES256, err := jwt.NewVerifierES(jwt.ES256, ecdsaPubKey); err != nil { if !errors.Is(err, jwt.ErrInvalidKey) { return nil, err } } else { alg.ES256 = verifierES256 algorithms = append(algorithms, "ES256") } if verifierES384, err := jwt.NewVerifierES(jwt.ES384, ecdsaPubKey); err != nil { if !errors.Is(err, jwt.ErrInvalidKey) { return nil, err } } else { alg.ES384 = verifierES384 algorithms = append(algorithms, "ES384") } if verifierES512, err := jwt.NewVerifierES(jwt.ES512, ecdsaPubKey); err != nil { if !errors.Is(err, jwt.ErrInvalidKey) { return nil, err } } else { alg.ES512 = verifierES512 algorithms = append(algorithms, "ES512") } } if len(algorithms) > 0 { log.Info().Str("algorithms", strings.Join(algorithms, ", ")).Msg("enabled JWT verifiers") } return alg, nil } func (s *algorithms) verify(token *jwt.Token) error { var verifier jwt.Verifier switch token.Header().Algorithm { case jwt.HS256: verifier = s.HS256 case jwt.HS384: verifier = s.HS384 case jwt.HS512: verifier = s.HS512 case jwt.RS256: verifier = s.RS256 case jwt.RS384: verifier = s.RS384 case jwt.RS512: verifier = s.RS512 case jwt.ES256: verifier = s.ES256 case jwt.ES384: verifier = s.ES384 case jwt.ES512: verifier = s.ES512 default: return fmt.Errorf("%w: %s", errUnsupportedAlgorithm, string(token.Header().Algorithm)) } if verifier == nil { return fmt.Errorf("%w: %s", errDisabledAlgorithm, string(token.Header().Algorithm)) } return verifier.Verify(token) } // verifySignature verifies a token's signature. The caller must hold // verifier.mu (at least RLock); it does not lock, to avoid recursive RLock. func (verifier *VerifierJWT) verifySignature(token *jwt.Token) error { err := verifier.algorithms.verify(token) if err != nil && verifier.previousHMACAlgorithms != nil { if validUntil := verifier.hmacPreviousSecretKeyValidUntil; validUntil > 0 && time.Now().Unix() > validUntil { return err } if prevErr := verifier.previousHMACAlgorithms.verify(token); prevErr == nil { return nil } } return err } // verifySignatureByJWK verifies a token against the JWKS manager. The caller // must hold verifier.mu (at least RLock); it does not lock, to avoid recursive // RLock. func (verifier *VerifierJWT) verifySignatureByJWK(token *jwt.Token, tokenVars map[string]any) error { return verifier.jwksManager.verify(token, tokenVars) } // extractTokenVars extracts named group matches from issuer/audience regex into tokenVars. // This must be called before signature verification because the extracted values may be // needed to construct the JWKS endpoint URL. Returns an error if regex is configured but // the claim doesn't match. func (verifier *VerifierJWT) extractTokenVars(claims jwt.RegisteredClaims, tokenVars map[string]any) error { if verifier.issuerRe != nil { match := verifier.issuerRe.FindStringSubmatch(claims.Issuer) if len(match) == 0 { return errors.New("issuer not matched") } for i, name := range verifier.issuerRe.SubexpNames() { if i != 0 && name != "" { tokenVars[name] = match[i] } } } if verifier.audienceRe != nil { matched := false for _, aud := range claims.Audience { match := verifier.audienceRe.FindStringSubmatch(aud) if len(match) == 0 { continue } matched = true for i, name := range verifier.audienceRe.SubexpNames() { if i != 0 && name != "" { tokenVars[name] = match[i] } } break } if !matched { return errors.New("audience not matched") } } return nil } // validateClaims checks issuer and audience claims against configured values. // Must be called after signature verification. func (verifier *VerifierJWT) validateClaims(claims jwt.RegisteredClaims) error { if verifier.audience != "" && !claims.IsForAudience(verifier.audience) { return errors.New("invalid audience") } if verifier.issuer != "" && !claims.IsIssuer(verifier.issuer) { return errors.New("invalid issuer") } return nil } func (verifier *VerifierJWT) VerifyConnectToken(t string, skipVerify bool) (ConnectToken, error) { token, err := jwt.ParseNoVerify([]byte(t)) // Will be verified later. if err != nil { return ConnectToken{}, fmt.Errorf("%w: %v", ErrInvalidToken, err) } claims, err := claimsDecoder.DecodeConnectClaims(token.Claims()) if err != nil { return ConnectToken{}, fmt.Errorf("%w: %v", ErrInvalidToken, err) } // Hold a read lock across the whole verification so a concurrent Reload (which // rewrites the verifier's signing/config fields under mu.Lock) can't change // them mid-verification - reading them unlocked was a data race and could // nil-deref algorithms/jwksManager when a reload switched auth mode. The // signature helpers assume the caller holds this lock and don't re-acquire it // (recursive RLock can deadlock if a writer is waiting). verifier.mu.RLock() defer verifier.mu.RUnlock() tokenVars := map[string]any{} // Extract token vars before signature verification — needed to construct JWKS URL. if err = verifier.extractTokenVars(claims.RegisteredClaims, tokenVars); err != nil { return ConnectToken{}, fmt.Errorf("%w: %v", ErrInvalidToken, err) } if !skipVerify { if verifier.jwksManager != nil { err = verifier.verifySignatureByJWK(token, tokenVars) } else { err = verifier.verifySignature(token) } if err != nil { return ConnectToken{}, fmt.Errorf("%w: %v", ErrInvalidToken, err) } } // Validate claims after signature verification. if err = verifier.validateClaims(claims.RegisteredClaims); err != nil { return ConnectToken{}, fmt.Errorf("%w: %v", ErrInvalidToken, err) } if claims.Channel != "" { return ConnectToken{}, fmt.Errorf( "%w: connection JWT can not contain channel claim, only subscription JWT can", ErrInvalidToken) } now := time.Now() if !claims.IsValidExpiresAt(now) || !claims.IsValidNotBefore(now) { return ConnectToken{}, ErrTokenExpired } subs := map[string]centrifuge.SubscribeOptions{} if len(claims.Subs) > 0 { for ch, v := range claims.Subs { _, _, chOpts, found, err := verifier.cfgContainer.ChannelOptions(ch) if err != nil { return ConnectToken{}, err } if !found { return ConnectToken{}, centrifuge.ErrorUnknownChannel } var info []byte if v.Base64Info != "" { byteInfo, err := base64.StdEncoding.DecodeString(v.Base64Info) if err != nil { return ConnectToken{}, fmt.Errorf("%w: %v", ErrInvalidToken, err) } info = byteInfo } else { info = v.Info } var data []byte if v.Base64Data != "" { byteInfo, err := base64.StdEncoding.DecodeString(v.Base64Data) if err != nil { return ConnectToken{}, fmt.Errorf("%w: %v", ErrInvalidToken, err) } data = byteInfo } else { data = v.Data } presence := chOpts.Presence if v.Override != nil && v.Override.Presence != nil { presence = v.Override.Presence.Value } joinLeave := chOpts.JoinLeave if v.Override != nil && v.Override.JoinLeave != nil { joinLeave = v.Override.JoinLeave.Value } pushJoinLeave := chOpts.ForcePushJoinLeave if v.Override != nil && v.Override.ForcePushJoinLeave != nil { pushJoinLeave = v.Override.ForcePushJoinLeave.Value } recovery := chOpts.ForceRecovery if v.Override != nil && v.Override.ForceRecovery != nil { recovery = v.Override.ForceRecovery.Value } positioning := chOpts.ForcePositioning if v.Override != nil && v.Override.ForcePositioning != nil { positioning = v.Override.ForcePositioning.Value } recoveryMode := chOpts.GetRecoveryMode() subs[ch] = centrifuge.SubscribeOptions{ ChannelInfo: info, EmitPresence: presence, EmitJoinLeave: joinLeave, PushJoinLeave: pushJoinLeave, EnableRecovery: recovery, EnablePositioning: positioning, RecoveryMode: recoveryMode, Data: data, Source: subsource.ConnectionToken, HistoryMetaTTL: chOpts.HistoryMetaTTL.ToDuration(), AllowedDeltaTypes: chOpts.AllowedDeltaTypes, AllowTagsFilter: chOpts.AllowTagsFilter, } } } else if len(claims.Channels) > 0 { for _, ch := range claims.Channels { _, _, chOpts, found, err := verifier.cfgContainer.ChannelOptions(ch) if err != nil { return ConnectToken{}, err } if !found { return ConnectToken{}, centrifuge.ErrorUnknownChannel } subs[ch] = centrifuge.SubscribeOptions{ EmitPresence: chOpts.Presence, EmitJoinLeave: chOpts.JoinLeave, PushJoinLeave: chOpts.ForcePushJoinLeave, EnableRecovery: chOpts.ForceRecovery, EnablePositioning: chOpts.ForcePositioning, RecoveryMode: chOpts.GetRecoveryMode(), Source: subsource.ConnectionToken, HistoryMetaTTL: chOpts.HistoryMetaTTL.ToDuration(), AllowedDeltaTypes: chOpts.AllowedDeltaTypes, AllowTagsFilter: chOpts.AllowTagsFilter, } } } var expireAt int64 if claims.ExpireAt != nil { if *claims.ExpireAt > 0 { expireAt = *claims.ExpireAt } } else { if claims.ExpiresAt != nil { expireAt = claims.ExpiresAt.Unix() } } var info []byte if claims.Base64Info != "" { byteInfo, err := base64.StdEncoding.DecodeString(claims.Base64Info) if err != nil { return ConnectToken{}, err } info = byteInfo } else { info = claims.Info } ct := ConnectToken{ Info: info, Subs: subs, ExpireAt: expireAt, Meta: claims.Meta, } if verifier.userIDClaim != "" { value := gjson.GetBytes(token.Claims(), verifier.userIDClaim) ct.UserID = value.String() } else { ct.UserID = claims.RegisteredClaims.Subject } return ct, nil } func (verifier *VerifierJWT) VerifySubscribeToken(t string, skipVerify bool) (SubscribeToken, error) { token, err := jwt.ParseNoVerify([]byte(t)) // Will be verified later. if err != nil { return SubscribeToken{}, fmt.Errorf("%w: %v", ErrInvalidToken, err) } claims, err := claimsDecoder.DecodeSubscribeClaims(token.Claims()) if err != nil { return SubscribeToken{}, fmt.Errorf("%w: %v", ErrInvalidToken, err) } // See VerifyConnectToken: hold a read lock across verification so a concurrent // Reload can't change the signing/config fields mid-verification. verifier.mu.RLock() defer verifier.mu.RUnlock() tokenVars := map[string]any{} // Extract token vars before signature verification — needed to construct JWKS URL. if err = verifier.extractTokenVars(claims.RegisteredClaims, tokenVars); err != nil { return SubscribeToken{}, fmt.Errorf("%w: %v", ErrInvalidToken, err) } if !skipVerify { if verifier.jwksManager != nil { err = verifier.verifySignatureByJWK(token, tokenVars) } else { err = verifier.verifySignature(token) } if err != nil { return SubscribeToken{}, fmt.Errorf("%w: %v", ErrInvalidToken, err) } } // Validate claims after signature verification. if err = verifier.validateClaims(claims.RegisteredClaims); err != nil { return SubscribeToken{}, fmt.Errorf("%w: %v", ErrInvalidToken, err) } now := time.Now() if !claims.IsValidExpiresAt(now) || !claims.IsValidNotBefore(now) { return SubscribeToken{}, ErrTokenExpired } if claims.Channel == "" { return SubscribeToken{}, fmt.Errorf("%w: channel claim is required for subscription JWT", ErrInvalidToken) } _, _, chOpts, found, err := verifier.cfgContainer.ChannelOptions(claims.Channel) if err != nil { return SubscribeToken{}, err } if !found { return SubscribeToken{}, centrifuge.ErrorUnknownChannel } var info []byte if claims.Base64Info != "" { byteInfo, err := base64.StdEncoding.DecodeString(claims.Base64Info) if err != nil { return SubscribeToken{}, fmt.Errorf("%w: %v", ErrInvalidToken, err) } info = byteInfo } else { info = claims.Info } var data []byte if claims.Base64Data != "" { byteInfo, err := base64.StdEncoding.DecodeString(claims.Base64Data) if err != nil { return SubscribeToken{}, fmt.Errorf("%w: %v", ErrInvalidToken, err) } data = byteInfo } else { data = claims.Data } presence := chOpts.Presence if claims.Override != nil && claims.Override.Presence != nil { presence = claims.Override.Presence.Value } joinLeave := chOpts.JoinLeave if claims.Override != nil && claims.Override.JoinLeave != nil { joinLeave = claims.Override.JoinLeave.Value } pushJoinLeave := chOpts.ForcePushJoinLeave if claims.Override != nil && claims.Override.ForcePushJoinLeave != nil { pushJoinLeave = claims.Override.ForcePushJoinLeave.Value } recovery := chOpts.ForceRecovery if claims.Override != nil && claims.Override.ForceRecovery != nil { recovery = claims.Override.ForceRecovery.Value } positioning := chOpts.ForcePositioning if claims.Override != nil && claims.Override.ForcePositioning != nil { positioning = claims.Override.ForcePositioning.Value } recoveryMode := chOpts.GetRecoveryMode() var expireAt int64 if claims.ExpireAt != nil { if *claims.ExpireAt > 0 { expireAt = *claims.ExpireAt } } else { if claims.ExpiresAt != nil { expireAt = claims.ExpiresAt.Unix() } } st := SubscribeToken{ UserID: claims.RegisteredClaims.Subject, Channel: claims.Channel, Client: claims.Client, Options: centrifuge.SubscribeOptions{ ExpireAt: expireAt, ChannelInfo: info, EmitPresence: presence, EmitJoinLeave: joinLeave, PushJoinLeave: pushJoinLeave, EnableRecovery: recovery, EnablePositioning: positioning, RecoveryMode: recoveryMode, AllowedDeltaTypes: chOpts.AllowedDeltaTypes, AllowTagsFilter: chOpts.AllowTagsFilter, Data: data, }, } if verifier.userIDClaim != "" { value := gjson.GetBytes(token.Claims(), verifier.userIDClaim) st.UserID = value.String() } else { st.UserID = claims.RegisteredClaims.Subject } return st, nil } func (verifier *VerifierJWT) Reload(config VerifierConfig) error { if err := config.Validate(); err != nil { return fmt.Errorf("error validating token verifier config: %w", err) } verifier.mu.Lock() defer verifier.mu.Unlock() var audienceRe *regexp.Regexp var issuerRe *regexp.Regexp if config.AudienceRegex != "" { var err error audienceRe, err = regexp.Compile(config.AudienceRegex) if err != nil { return fmt.Errorf("error compiling audience regex: %w", err) } } if config.IssuerRegex != "" { var err error issuerRe, err = regexp.Compile(config.IssuerRegex) if err != nil { return fmt.Errorf("error compiling issuer regex: %w", err) } } if config.JWKSPublicEndpoint != "" { mng, err := jwks.NewManager(config.JWKSPublicEndpoint) if err != nil { return fmt.Errorf("error creating JWK manager: %w", err) } verifier.jwksManager = &jwksManager{mng} verifier.algorithms = nil verifier.previousHMACAlgorithms = nil verifier.hmacPreviousSecretKeyValidUntil = 0 } else { alg, err := newAlgorithms(config.HMACSecretKey, config.RSAPublicKey, config.ECDSAPublicKey) if err != nil { return err } verifier.algorithms = alg verifier.jwksManager = nil if config.HMACPreviousSecretKey != "" { prevAlg, err := newAlgorithms(config.HMACPreviousSecretKey, nil, nil) if err != nil { return fmt.Errorf("error initializing previous HMAC token algorithms: %w", err) } verifier.previousHMACAlgorithms = prevAlg verifier.hmacPreviousSecretKeyValidUntil = config.HMACPreviousSecretKeyValidUntil } else { verifier.previousHMACAlgorithms = nil verifier.hmacPreviousSecretKeyValidUntil = 0 } } verifier.audience = config.Audience verifier.audienceRe = audienceRe verifier.issuer = config.Issuer verifier.issuerRe = issuerRe verifier.userIDClaim = config.UserIDClaim return nil }