diff --git a/.pubnub.yml b/.pubnub.yml index 0456cddf..1ed77d4a 100644 --- a/.pubnub.yml +++ b/.pubnub.yml @@ -1,6 +1,21 @@ --- -version: v9.0.2 +version: v9.0.3 changelog: + - date: 2026-07-02 + version: v9.0.3 + changes: + - type: bug + text: "Add `recover` guards to SDK-owned goroutines (subscribe worker, request workers, job enqueue) so panics fail the request instead of crashing the host process." + - type: bug + text: "Replace panic-based `CheckUUID` with `ValidateUUID` and validate the UUID in `buildURL`, surfacing invalid UUIDs as handled errors; `CheckUUID` is kept but deprecated." + - type: bug + text: "Harden subscribe response parsing against malformed payloads with checked type assertions that announce a parse error instead of panicking." + - type: bug + text: "Harden Access Manager grant response parsing to validate all fields and return descriptive errors, and fix TTL parsing to accept JSON numbers." + - type: improvement + text: "Return a single generic error on LegacyCryptor decryption failures to prevent padding-oracle-style information disclosure." + - type: improvement + text: "Change redaction of `SecretKey` to show only first 8 characters and total length of the key." - date: 2026-06-23 version: v9.0.2 changes: @@ -849,7 +864,7 @@ sdks: distribution-type: package distribution-repository: GitHub package-name: Go - location: https://github.com/pubnub/go/releases/tag/v9.0.2 + location: https://github.com/pubnub/go/releases/tag/v9.0.3 requires: - name: "Go" diff --git a/CHANGELOG.md b/CHANGELOG.md index 8bb9c212..2a8ac7f4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,16 @@ +## v9.0.3 +July 02 2026 + +#### Fixed +- Add `recover` guards to SDK-owned goroutines (subscribe worker, request workers, job enqueue) so panics fail the request instead of crashing the host process. +- Replace panic-based `CheckUUID` with `ValidateUUID` and validate the UUID in `buildURL`, surfacing invalid UUIDs as handled errors; `CheckUUID` is kept but deprecated. +- Harden subscribe response parsing against malformed payloads with checked type assertions that announce a parse error instead of panicking. +- Harden Access Manager grant response parsing to validate all fields and return descriptive errors, and fix TTL parsing to accept JSON numbers. + +#### Modified +- Return a single generic error on LegacyCryptor decryption failures to prevent padding-oracle-style information disclosure. +- Change redaction of `SecretKey` to show only first 8 characters and total length of the key. + ## v9.0.2 June 23 2026 diff --git a/VERSION b/VERSION index 3beeadd4..66e3058d 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -9.0.2 +9.0.3 diff --git a/config.go b/config.go index 1c214cad..c449f8b1 100644 --- a/config.go +++ b/config.go @@ -149,8 +149,28 @@ func (c *Config) GetUserId() UserId { return UserId(c.UUID) } +// secretKeyRedactPrefixLen is the number of leading characters preserved when +// partially redacting a SecretKey for logging (e.g. "sec-c-YT… (len=54)"). +const secretKeyRedactPrefixLen = 8 + +// redactPrefix returns a partially redacted representation of a sensitive value: +// the first prefixLen characters followed by an ellipsis and the total length, +// e.g. "sec-c-YT… (len=54)". +// +// An empty value yields an empty string. To avoid leaking short secrets, any +// value whose length does not exceed prefixLen is fully masked with "***". +func redactPrefix(value string, prefixLen int) string { + if value == "" { + return "" + } + if len(value) <= prefixLen { + return "***" + } + return fmt.Sprintf("%s\u2026 (len=%d)", value[:prefixLen], len(value)) +} + // GetLogString returns a formatted string representation of the Config for logging purposes. -// Sensitive fields (SecretKey, CipherKey) are masked with ***. +// Sensitive fields (SecretKey, CipherKey) are masked. func (c *Config) GetLogString() string { c.RLock() defer c.RUnlock() @@ -203,7 +223,7 @@ func (c *Config) GetLogString() string { }`, c.PublishKey, c.SubscribeKey, - maskIfNotEmpty(c.SecretKey), + redactPrefix(c.SecretKey, secretKeyRedactPrefixLen), maskIfNotEmpty(c.AuthKey), c.Origin, c.UUID, diff --git a/crypto/legacy_cryptor.go b/crypto/legacy_cryptor.go index cf679fa0..85bb32de 100644 --- a/crypto/legacy_cryptor.go +++ b/crypto/legacy_cryptor.go @@ -8,7 +8,6 @@ import ( "crypto/sha256" "encoding/hex" "errors" - "fmt" "io" ) @@ -58,12 +57,18 @@ func (c *legacyCryptor) Encrypt(message []byte) (*EncryptedData, error) { } func (c *legacyCryptor) Decrypt(encryptedData *EncryptedData) (r []byte, e error) { + defer func() { + if rec := recover(); rec != nil { + r, e = nil, errors.New("decryption error") + } + }() + iv := make([]byte, aes.BlockSize) data := encryptedData.Data if c.randomIv { if len(data) < aes.BlockSize { - return nil, fmt.Errorf("length of data to decrypt should be at least %d", aes.BlockSize) + return nil, errors.New("decryption error") } iv = data[:aes.BlockSize] @@ -72,26 +77,14 @@ func (c *legacyCryptor) Decrypt(encryptedData *EncryptedData) (r []byte, e error iv = []byte(valIV) } - if len(data)%16 != 0 { - return nil, fmt.Errorf("length of data to decrypt should be divisible by block size") + if len(data)%aes.BlockSize != 0 { + return nil, errors.New("decryption error") } decrypter := cipher.NewCBCDecrypter(c.block, iv) - //to handle decryption errors - defer func() { - if rec := recover(); rec != nil { - r, e = nil, fmt.Errorf("decrypt error: %s", rec) - } - }() - decrypted := make([]byte, len(data)) decrypter.CryptBlocks(decrypted, data) - val, err := unpadPKCS7(decrypted) - if err != nil { - return nil, fmt.Errorf("decrypt error: %s", err) - } - - return val, nil + return unpadPKCS7(decrypted) } func (c *legacyCryptor) EncryptStream(reader io.Reader) (*EncryptedStreamData, error) { diff --git a/endpoints.go b/endpoints.go index 885c496c..f464d4a3 100644 --- a/endpoints.go +++ b/endpoints.go @@ -137,7 +137,6 @@ func defaultQuery(uuid string, telemetryManager *TelemetryManager) *url.Values { v := &url.Values{} v.Set("pnsdk", "PubNub-Go/"+Version) - utils.CheckUUID(uuid) v.Set("uuid", uuid) for queryName, queryParam := range telemetryManager.OperationLatency() { @@ -161,6 +160,15 @@ func buildURL(o endpoint) (*url.URL, error) { return &url.URL{}, err } + // Validate the client UUID for every endpoint that sends it (those built + // via defaultQuery). Returning an error here lets callers handle an invalid + // UUID gracefully instead of panicking inside SDK-owned goroutines. + if query.Has("uuid") { + if err := utils.ValidateUUID(query.Get("uuid")); err != nil { + return &url.URL{}, err + } + } + if v := o.tokenManager().GetToken(); v != "" && query.Get("auth") == "" { query.Set("auth", v) } else if v := o.config().AuthKey; v != "" && query.Get("auth") == "" { diff --git a/enums.go b/enums.go index 4fbe641d..96e80121 100644 --- a/enums.go +++ b/enums.go @@ -48,9 +48,17 @@ type PNPushEnvironment string // PNLogLevel constants define the available log levels const ( - // PNLogLevelTrace is the enum when the log level is trace + // PNLogLevelTrace is the enum when the log level is trace. + // + // Security warning: like Debug, the Trace level may expose sensitive fields + // (tokens, auth parameters, request URLs) and must not be enabled in + // production environments. See PNLogLevelDebug for details. PNLogLevelTrace PNLogLevel = 1 + iota - // PNLogLevelDebug is the enum when the log level is debug + // PNLogLevelDebug is the enum when the log level is debug. + // + // Security warning: the Debug log level may expose sensitive fields (tokens, + // auth parameters, request URLs). Debug is intended for developer + // troubleshooting only and must not be enabled in production environments. PNLogLevelDebug // PNLogLevelInfo is the enum when the log level is info PNLogLevelInfo diff --git a/grant_request.go b/grant_request.go index 3f096738..71421734 100644 --- a/grant_request.go +++ b/grant_request.go @@ -355,114 +355,156 @@ func newGrantResponse(jsonBytes []byte, status StatusResponse) ( grantData, ok := value.(map[string]interface{}) if !ok { - e := pnerr.NewResponseParsingError("Error parsing response: invalid JSON structure", - io.NopCloser(bytes.NewBufferString(string(jsonBytes))), + e := newGrantResponseParsingError(jsonBytes, "invalid JSON structure", fmt.Errorf("expected map[string]interface{}, got %T", value)) return emptyGrantResponse, status, e } payload, ok := grantData["payload"] if !ok { - e := pnerr.NewResponseParsingError("Error parsing response: missing payload field", - io.NopCloser(bytes.NewBufferString(string(jsonBytes))), + e := newGrantResponseParsingError(jsonBytes, "missing payload field", fmt.Errorf("payload field not found in response")) return emptyGrantResponse, status, e } if payload == nil { - e := pnerr.NewResponseParsingError("Error parsing response: null payload", - io.NopCloser(bytes.NewBufferString(string(jsonBytes))), + e := newGrantResponseParsingError(jsonBytes, "null payload", fmt.Errorf("payload field is null")) return emptyGrantResponse, status, e } parsedPayload, ok := payload.(map[string]interface{}) if !ok { - e := pnerr.NewResponseParsingError("Error parsing response: invalid payload structure", - io.NopCloser(bytes.NewBufferString(string(jsonBytes))), + e := newGrantResponseParsingError(jsonBytes, "invalid payload structure", fmt.Errorf("expected map[string]interface{} for payload, got %T", payload)) return emptyGrantResponse, status, e } - auths, _ := parsedPayload["auths"].(map[string]interface{}) - ttl, _ := parsedPayload["ttl"].(float64) + rootAuths, _, err := grantOptionalMap(parsedPayload, "auths") + if err != nil { + return emptyGrantResponse, status, newGrantResponseParsingError(jsonBytes, "invalid auths field", err) + } + ttl, _, err := grantOptionalNumber(parsedPayload, "ttl") + if err != nil { + return emptyGrantResponse, status, newGrantResponseParsingError(jsonBytes, "invalid ttl field", err) + } if val, ok := parsedPayload["channel"]; ok { - channelName := val.(string) - auths := make(map[string]*PNAccessManagerKeyData) - channelMap, _ := parsedPayload["auths"].(map[string]interface{}) + channelName, ok := val.(string) + if !ok { + return emptyGrantResponse, status, newGrantResponseParsingError(jsonBytes, "invalid channel field", + fmt.Errorf("expected string for channel, got %T", val)) + } + constructedAuthKeys := make(map[string]*PNAccessManagerKeyData) entityData := &PNPAMEntityData{ Name: channelName, } - for key, value := range channelMap { - auths[key] = createPNAccessManagerKeyData(value, entityData, false) + for key, value := range rootAuths { + keyData, err := createPNAccessManagerKeyData(value, entityData, false) + if err != nil { + return emptyGrantResponse, status, newGrantResponseParsingError(jsonBytes, fmt.Sprintf("invalid auth key %q", key), err) + } + constructedAuthKeys[key] = keyData } - entityData.AuthKeys = auths + entityData.AuthKeys = constructedAuthKeys entityData.TTL = int(ttl) constructedChannels[channelName] = entityData } if val, ok := parsedPayload["channel-groups"]; ok { - groupName, _ := val.(string) constructedAuthKey := make(map[string]*PNAccessManagerKeyData) entityData := &PNPAMEntityData{ - Name: groupName, + Name: "", } - if _, ok := val.(string); ok { - for authKeyName, value := range auths { - constructedAuthKey[authKeyName] = createPNAccessManagerKeyData(value, entityData, false) + if groupName, ok := val.(string); ok { + entityData.Name = groupName + for authKeyName, value := range rootAuths { + keyData, err := createPNAccessManagerKeyData(value, entityData, false) + if err != nil { + return emptyGrantResponse, status, newGrantResponseParsingError(jsonBytes, fmt.Sprintf("invalid auth key %q", authKeyName), err) + } + constructedAuthKey[authKeyName] = keyData } entityData.AuthKeys = constructedAuthKey constructedGroups[groupName] = entityData - } - - if groupMap, ok := val.(map[string]interface{}); ok { - groupName, _ := val.(string) - constructedAuthKey := make(map[string]*PNAccessManagerKeyData) - entityData := &PNPAMEntityData{ - Name: groupName, - } - + } else if groupMap, ok := val.(map[string]interface{}); ok { for groupName, value := range groupMap { - valueMap := value.(map[string]interface{}) + valueMap, ok := value.(map[string]interface{}) + if !ok { + return emptyGrantResponse, status, newGrantResponseParsingError(jsonBytes, fmt.Sprintf("invalid channel group %q", groupName), + fmt.Errorf("expected map[string]interface{}, got %T", value)) + } if keys, ok := valueMap["auths"]; ok { - parsedKeys, _ := keys.(map[string]interface{}) + parsedKeys, ok := keys.(map[string]interface{}) + if !ok { + return emptyGrantResponse, status, newGrantResponseParsingError(jsonBytes, fmt.Sprintf("invalid auths for channel group %q", groupName), + fmt.Errorf("expected map[string]interface{}, got %T", keys)) + } for keyName, value := range parsedKeys { - constructedAuthKey[keyName] = createPNAccessManagerKeyData(value, entityData, false) + keyData, err := createPNAccessManagerKeyData(value, entityData, false) + if err != nil { + return emptyGrantResponse, status, newGrantResponseParsingError(jsonBytes, fmt.Sprintf("invalid auth key %q", keyName), err) + } + constructedAuthKey[keyName] = keyData } } - createPNAccessManagerKeyData(valueMap, entityData, true) + if _, err := createPNAccessManagerKeyData(valueMap, entityData, true); err != nil { + return emptyGrantResponse, status, newGrantResponseParsingError(jsonBytes, fmt.Sprintf("invalid channel group %q permissions", groupName), err) + } if val, ok := parsedPayload["ttl"]; ok { - parsedVal, _ := val.(float64) + parsedVal, ok := grantNumber(val) + if !ok { + return emptyGrantResponse, status, newGrantResponseParsingError(jsonBytes, "invalid ttl field", + fmt.Errorf("expected number for ttl, got %T", val)) + } entityData.TTL = int(parsedVal) } entityData.AuthKeys = constructedAuthKey constructedGroups[groupName] = entityData } + } else { + return emptyGrantResponse, status, newGrantResponseParsingError(jsonBytes, "invalid channel-groups field", + fmt.Errorf("expected string or map[string]interface{}, got %T", val)) } } if val, ok := parsedPayload["channels"]; ok { - channelMap, _ := val.(map[string]interface{}) + channelMap, ok := val.(map[string]interface{}) + if !ok { + return emptyGrantResponse, status, newGrantResponseParsingError(jsonBytes, "invalid channels field", + fmt.Errorf("expected map[string]interface{}, got %T", val)) + } for channelName, value := range channelMap { - constructedChannels[channelName] = fetchChannel(channelName, value, parsedPayload) + channelData, err := fetchChannel(channelName, value, parsedPayload) + if err != nil { + return emptyGrantResponse, status, newGrantResponseParsingError(jsonBytes, fmt.Sprintf("invalid channel %q", channelName), err) + } + constructedChannels[channelName] = channelData } } if val, ok := parsedPayload["uuids"]; ok { - uuids, _ := val.(map[string]interface{}) + uuids, ok := val.(map[string]interface{}) + if !ok { + return emptyGrantResponse, status, newGrantResponseParsingError(jsonBytes, "invalid uuids field", + fmt.Errorf("expected map[string]interface{}, got %T", val)) + } for uuid, value := range uuids { - constructedUUIDs[uuid] = fetchChannel(uuid, value, parsedPayload) + uuidData, err := fetchChannel(uuid, value, parsedPayload) + if err != nil { + return emptyGrantResponse, status, newGrantResponseParsingError(jsonBytes, fmt.Sprintf("invalid uuid %q", uuid), err) + } + constructedUUIDs[uuid] = uuidData } } @@ -475,65 +517,148 @@ func newGrantResponse(jsonBytes []byte, status StatusResponse) ( resp.ChannelGroups = constructedGroups resp.UUIDs = constructedUUIDs - resp.ReadEnabled = parsePerms(parsedPayload, "r") - resp.WriteEnabled = parsePerms(parsedPayload, "w") - resp.ManageEnabled = parsePerms(parsedPayload, "m") - resp.DeleteEnabled = parsePerms(parsedPayload, "d") - resp.GetEnabled = parsePerms(parsedPayload, "g") - resp.UpdateEnabled = parsePerms(parsedPayload, "u") - resp.JoinEnabled = parsePerms(parsedPayload, "j") + if resp.ReadEnabled, err = parsePerms(parsedPayload, "r"); err != nil { + return emptyGrantResponse, status, newGrantResponseParsingError(jsonBytes, "invalid r permission", err) + } + if resp.WriteEnabled, err = parsePerms(parsedPayload, "w"); err != nil { + return emptyGrantResponse, status, newGrantResponseParsingError(jsonBytes, "invalid w permission", err) + } + if resp.ManageEnabled, err = parsePerms(parsedPayload, "m"); err != nil { + return emptyGrantResponse, status, newGrantResponseParsingError(jsonBytes, "invalid m permission", err) + } + if resp.DeleteEnabled, err = parsePerms(parsedPayload, "d"); err != nil { + return emptyGrantResponse, status, newGrantResponseParsingError(jsonBytes, "invalid d permission", err) + } + if resp.GetEnabled, err = parsePerms(parsedPayload, "g"); err != nil { + return emptyGrantResponse, status, newGrantResponseParsingError(jsonBytes, "invalid g permission", err) + } + if resp.UpdateEnabled, err = parsePerms(parsedPayload, "u"); err != nil { + return emptyGrantResponse, status, newGrantResponseParsingError(jsonBytes, "invalid u permission", err) + } + if resp.JoinEnabled, err = parsePerms(parsedPayload, "j"); err != nil { + return emptyGrantResponse, status, newGrantResponseParsingError(jsonBytes, "invalid j permission", err) + } if r, ok := parsedPayload["ttl"]; ok { - parsedValue, _ := r.(float64) + parsedValue, ok := grantNumber(r) + if !ok { + return emptyGrantResponse, status, newGrantResponseParsingError(jsonBytes, "invalid ttl field", + fmt.Errorf("expected number for ttl, got %T", r)) + } resp.TTL = int(parsedValue) } return resp, status, nil } -func parsePerms(parsedPayload map[string]interface{}, name string) bool { +func newGrantResponseParsingError(jsonBytes []byte, message string, err error) error { + return pnerr.NewResponseParsingError("Error parsing response: "+message, + io.NopCloser(bytes.NewBufferString(string(jsonBytes))), err) +} + +func grantNumber(value interface{}) (float64, bool) { + switch typedValue := value.(type) { + case float64: + return typedValue, true + case int: + return float64(typedValue), true + default: + return 0, false + } +} + +func grantOptionalNumber(parsedPayload map[string]interface{}, name string) (float64, bool, error) { + value, ok := parsedPayload[name] + if !ok { + return 0, false, nil + } + + parsedValue, ok := grantNumber(value) + if !ok { + return 0, true, fmt.Errorf("expected number for %s, got %T", name, value) + } + + return parsedValue, true, nil +} + +func grantOptionalMap(parsedPayload map[string]interface{}, name string) (map[string]interface{}, bool, error) { + value, ok := parsedPayload[name] + if !ok { + return nil, false, nil + } + + parsedValue, ok := value.(map[string]interface{}) + if !ok { + return nil, true, fmt.Errorf("expected map[string]interface{} for %s, got %T", name, value) + } + + return parsedValue, true, nil +} + +func parsePerms(parsedPayload map[string]interface{}, name string) (bool, error) { if r, ok := parsedPayload[name]; ok { - parsedValue, _ := r.(float64) + parsedValue, ok := grantNumber(r) + if !ok { + return false, fmt.Errorf("expected number for %s, got %T", name, r) + } if parsedValue == float64(1) { - return true + return true, nil } else { - return false + return false, nil } } - return false + return false, nil } -func fetchChannel(channelName string, value interface{}, parsedPayload map[string]interface{}) *PNPAMEntityData { +func fetchChannel(channelName string, value interface{}, parsedPayload map[string]interface{}) (*PNPAMEntityData, error) { auths := make(map[string]*PNAccessManagerKeyData) entityData := &PNPAMEntityData{ Name: channelName, } - valueMap, _ := value.(map[string]interface{}) + valueMap, ok := value.(map[string]interface{}) + if !ok { + return nil, fmt.Errorf("expected map[string]interface{}, got %T", value) + } if val, ok := valueMap["auths"]; ok { - parsedValue := val.(map[string]interface{}) + parsedValue, ok := val.(map[string]interface{}) + if !ok { + return nil, fmt.Errorf("expected map[string]interface{} for auths, got %T", val) + } for key, value := range parsedValue { - auths[key] = createPNAccessManagerKeyData(value, entityData, false) + keyData, err := createPNAccessManagerKeyData(value, entityData, false) + if err != nil { + return nil, fmt.Errorf("invalid auth key %q: %w", key, err) + } + auths[key] = keyData } } - createPNAccessManagerKeyData(value, entityData, true) + if _, err := createPNAccessManagerKeyData(value, entityData, true); err != nil { + return nil, err + } if val, ok := parsedPayload["ttl"]; ok { - parsedVal, _ := val.(float64) + parsedVal, ok := grantNumber(val) + if !ok { + return nil, fmt.Errorf("expected number for ttl, got %T", val) + } entityData.TTL = int(parsedVal) } entityData.AuthKeys = auths - return entityData + return entityData, nil } -func readKeyData(val interface{}, keyData *PNAccessManagerKeyData, entityData *PNPAMEntityData, writeToEntityData bool, grantType PNGrantType) { - parsedValue, _ := val.(float64) +func readKeyData(val interface{}, keyData *PNAccessManagerKeyData, entityData *PNPAMEntityData, writeToEntityData bool, grantType PNGrantType) error { + parsedValue, ok := grantNumber(val) + if !ok { + return fmt.Errorf("expected number for permission, got %T", val) + } readValue := false if parsedValue == float64(1) { readValue = true @@ -573,43 +698,64 @@ func readKeyData(val interface{}, keyData *PNAccessManagerKeyData, entityData *P keyData.JoinEnabled = readValue } } + return nil } -func createPNAccessManagerKeyData(value interface{}, entityData *PNPAMEntityData, writeToEntityData bool) *PNAccessManagerKeyData { - valueMap := value.(map[string]interface{}) +func createPNAccessManagerKeyData(value interface{}, entityData *PNPAMEntityData, writeToEntityData bool) (*PNAccessManagerKeyData, error) { + valueMap, ok := value.(map[string]interface{}) + if !ok { + return nil, fmt.Errorf("expected map[string]interface{}, got %T", value) + } keyData := &PNAccessManagerKeyData{} if val, ok := valueMap["r"]; ok { - readKeyData(val, keyData, entityData, writeToEntityData, PNReadEnabled) + if err := readKeyData(val, keyData, entityData, writeToEntityData, PNReadEnabled); err != nil { + return nil, fmt.Errorf("invalid r permission: %w", err) + } } if val, ok := valueMap["w"]; ok { - readKeyData(val, keyData, entityData, writeToEntityData, PNWriteEnabled) + if err := readKeyData(val, keyData, entityData, writeToEntityData, PNWriteEnabled); err != nil { + return nil, fmt.Errorf("invalid w permission: %w", err) + } } if val, ok := valueMap["m"]; ok { - readKeyData(val, keyData, entityData, writeToEntityData, PNManageEnabled) + if err := readKeyData(val, keyData, entityData, writeToEntityData, PNManageEnabled); err != nil { + return nil, fmt.Errorf("invalid m permission: %w", err) + } } if val, ok := valueMap["d"]; ok { - readKeyData(val, keyData, entityData, writeToEntityData, PNDeleteEnabled) + if err := readKeyData(val, keyData, entityData, writeToEntityData, PNDeleteEnabled); err != nil { + return nil, fmt.Errorf("invalid d permission: %w", err) + } } if val, ok := valueMap["g"]; ok { - readKeyData(val, keyData, entityData, writeToEntityData, PNGetEnabled) + if err := readKeyData(val, keyData, entityData, writeToEntityData, PNGetEnabled); err != nil { + return nil, fmt.Errorf("invalid g permission: %w", err) + } } if val, ok := valueMap["u"]; ok { - readKeyData(val, keyData, entityData, writeToEntityData, PNUpdateEnabled) + if err := readKeyData(val, keyData, entityData, writeToEntityData, PNUpdateEnabled); err != nil { + return nil, fmt.Errorf("invalid u permission: %w", err) + } } if val, ok := valueMap["j"]; ok { - readKeyData(val, keyData, entityData, writeToEntityData, PNJoinEnabled) + if err := readKeyData(val, keyData, entityData, writeToEntityData, PNJoinEnabled); err != nil { + return nil, fmt.Errorf("invalid j permission: %w", err) + } } if val, ok := valueMap["ttl"]; ok { - parsedVal, _ := val.(int) - entityData.TTL = parsedVal + parsedVal, ok := grantNumber(val) + if !ok { + return nil, fmt.Errorf("expected number for ttl, got %T", val) + } + entityData.TTL = int(parsedVal) } - return keyData + return keyData, nil } diff --git a/grant_request_test.go b/grant_request_test.go index becbebc4..1a4eaf95 100644 --- a/grant_request_test.go +++ b/grant_request_test.go @@ -433,6 +433,89 @@ func TestGrantErrorResponseParsing(t *testing.T) { assert.Contains(err.Error(), "null payload") } +func TestGrantNestedErrorResponseParsing(t *testing.T) { + tests := []struct { + name string + jsonBytes []byte + contains string + }{ + { + name: "channel is not string", + jsonBytes: []byte(`{"message":"Success","payload":{"ttl":1440,"channel":42,"auths":{"my-auth-key":{"r":1}}},"service":"Access Manager","status":200}`), + contains: "invalid channel field", + }, + { + name: "root auths is not map", + jsonBytes: []byte(`{"message":"Success","payload":{"ttl":1440,"channel":"ch1","auths":"bad-auths"},"service":"Access Manager","status":200}`), + contains: "invalid auths field", + }, + { + name: "auth key value is not map", + jsonBytes: []byte(`{"message":"Success","payload":{"ttl":1440,"channel":"ch1","auths":{"my-auth-key":null}},"service":"Access Manager","status":200}`), + contains: "invalid auth key", + }, + { + name: "channel groups value is not string or map", + jsonBytes: []byte(`{"message":"Success","payload":{"ttl":1440,"channel-groups":42},"service":"Access Manager","status":200}`), + contains: "invalid channel-groups field", + }, + { + name: "channel group entry is not map", + jsonBytes: []byte(`{"message":"Success","payload":{"ttl":1440,"channel-groups":{"cg1":null}},"service":"Access Manager","status":200}`), + contains: "invalid channel group", + }, + { + name: "channel group auths is not map", + jsonBytes: []byte(`{"message":"Success","payload":{"ttl":1440,"channel-groups":{"cg1":{"auths":null}}},"service":"Access Manager","status":200}`), + contains: "invalid auths for channel group", + }, + { + name: "channels field is not map", + jsonBytes: []byte(`{"message":"Success","payload":{"ttl":1440,"channels":42},"service":"Access Manager","status":200}`), + contains: "invalid channels field", + }, + { + name: "channel entry is not map", + jsonBytes: []byte(`{"message":"Success","payload":{"ttl":1440,"channels":{"ch1":null}},"service":"Access Manager","status":200}`), + contains: "invalid channel", + }, + { + name: "channel auths is not map", + jsonBytes: []byte(`{"message":"Success","payload":{"ttl":1440,"channels":{"ch1":{"auths":null}}},"service":"Access Manager","status":200}`), + contains: "invalid channel", + }, + { + name: "uuids field is not map", + jsonBytes: []byte(`{"message":"Success","payload":{"ttl":1440,"uuids":42},"service":"Access Manager","status":200}`), + contains: "invalid uuids field", + }, + { + name: "uuid auth permission is not number", + jsonBytes: []byte(`{"message":"Success","payload":{"ttl":1440,"uuids":{"uuid1":{"auths":{"my-auth-key":{"r":"bad"}}}}},"service":"Access Manager","status":200}`), + contains: "invalid uuid", + }, + { + name: "root permission is not number", + jsonBytes: []byte(`{"message":"Success","payload":{"ttl":1440,"r":"bad"},"service":"Access Manager","status":200}`), + contains: "invalid r permission", + }, + { + name: "ttl is not number", + jsonBytes: []byte(`{"message":"Success","payload":{"ttl":"bad"},"service":"Access Manager","status":200}`), + contains: "invalid ttl field", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, _, err := newGrantResponse(tt.jsonBytes, StatusResponse{}) + assert.NotNil(t, err) + assert.Contains(t, err.Error(), "pubnub/parsing") + assert.Contains(t, err.Error(), tt.contains) + }) + } +} + func TestGrantSpecialCharactersInIdentifiers(t *testing.T) { assert := assert.New(t) pn := NewPubNub(NewDemoConfig()) diff --git a/pubnub.go b/pubnub.go index c25fb2d9..cbb5c052 100644 --- a/pubnub.go +++ b/pubnub.go @@ -18,7 +18,7 @@ import ( // Default constants const ( // Version :the version of the SDK - Version = "9.0.2" + Version = "9.0.3" // MaxSequence for publish messages MaxSequence = 65535 ) diff --git a/request.go b/request.go index d5f348a0..3a63b8dc 100644 --- a/request.go +++ b/request.go @@ -61,6 +61,16 @@ func fillJobQ(req *http.Request, client *http.Client, opts endpoint, j chan *Job } func addToJobQ(req *http.Request, client *http.Client, opts endpoint, j chan *JobQResponse, ctx Context) { + defer func() { + if r := recover(); r != nil { + err := fmt.Errorf("request enqueue panic: %v", r) + opts.getPubNub().loggerManager.LogSimple(PNLogLevelError, err.Error(), false) + // Release the caller waiting on j so the request fails with an + // error instead of blocking forever after the panic. + j <- &JobQResponse{Error: err} + } + }() + if ctx != nil { select { case <-ctx.Done(): diff --git a/request_workers.go b/request_workers.go index 04e1a748..ad04ad17 100644 --- a/request_workers.go +++ b/request_workers.go @@ -53,6 +53,12 @@ func newRequestWorkers(workers chan chan *JobQItem, id int, ctx Context) Worker // Process runs a goroutine for the worker func (pw Worker) Process(pubnub *PubNub) { go func() { + defer func() { + if r := recover(); r != nil { + pubnub.loggerManager.LogSimple(PNLogLevelError, + fmt.Sprintf("Recovered from panic in worker Process, id %d: %v", pw.id, r), false) + } + }() ProcessLabel: for { select { diff --git a/subscription_manager.go b/subscription_manager.go index a18a2b4a..46c2890f 100644 --- a/subscription_manager.go +++ b/subscription_manager.go @@ -510,6 +510,13 @@ type originationMetadata struct { } func subscribeMessageWorker(m *SubscriptionManager) { + defer func() { + if rec := recover(); rec != nil { + err := fmt.Errorf("subscribeMessageWorker panic: %v", rec) + m.pubnub.loggerManager.LogError(err, "SubscribeMessageWorkerPanic", PNSubscribeOperation, true) + } + }() + m.Lock() if m.ctx == nil && m.subscribeCancel == nil { m.pubnub.loggerManager.LogSimple(PNLogLevelTrace, "subscribeMessageWorker: setting context", false) @@ -542,13 +549,89 @@ SubscribeMessageWorkerLabel: break SubscribeMessageWorkerLabel case message := <-m.messages: m.pubnub.loggerManager.LogSimple(PNLogLevelTrace, "subscribeMessageWorker: processing message", false) - processSubscribePayload(m, message) + safeProcessSubscribePayload(m, message) } } m.pubnub.loggerManager.LogSimple(PNLogLevelTrace, "subscribeMessageWorker: exited", false) } +func announceSubscribePayloadParsingError(m *SubscriptionManager, channel, message string) { + err := errors.New(message) + m.pubnub.loggerManager.LogError(err, "SubscribePayloadParsingFailed", PNSubscribeOperation, true) + m.listenerManager.announceStatus(&PNStatus{ + Category: PNUnknownCategory, + ErrorData: err, + Error: true, + Operation: PNSubscribeOperation, + AffectedChannels: []string{channel}, + }) +} + +func safeProcessSubscribePayload(m *SubscriptionManager, payload subscribeMessage) { + defer func() { + if rec := recover(); rec != nil { + err := fmt.Errorf("Subscribe payload processing panic: %v", rec) + m.pubnub.loggerManager.LogError(err, "SubscribePayloadProcessingPanic", PNSubscribeOperation, true) + m.listenerManager.announceStatus(&PNStatus{ + Category: PNUnknownCategory, + ErrorData: err, + Error: true, + Operation: PNSubscribeOperation, + AffectedChannels: []string{payload.Channel}, + }) + } + }() + + processSubscribePayload(m, payload) +} + +func subscribePayloadMap(value interface{}) (map[string]interface{}, bool) { + payload, ok := value.(map[string]interface{}) + if !ok { + return nil, false + } + + return payload, true +} + +func subscribePayloadRequiredString(payload map[string]interface{}, field string) (string, bool) { + value, ok := payload[field] + if !ok { + return "", false + } + + str, ok := value.(string) + if !ok { + return "", false + } + + return str, true +} + +func subscribePayloadOptionalString(payload map[string]interface{}, field string) (string, bool) { + value, ok := payload[field] + if !ok { + return "", true + } + + str, ok := value.(string) + if !ok { + return "", false + } + + return str, true +} + +func subscribePayloadOptionalMap(payload map[string]interface{}, field string) (map[string]interface{}, bool) { + value, ok := payload[field] + if !ok { + return nil, true + } + + return subscribePayloadMap(value) +} + func processPresencePayload(m *SubscriptionManager, payload subscribeMessage, channel, subscriptionMatch string, publishMeta publishMetadata) { var presencePayload map[string]interface{} var action, uuid, actualChannel, subscribedChannel string @@ -558,13 +641,8 @@ func processPresencePayload(m *SubscriptionManager, payload subscribeMessage, ch var ok, hereNowRefresh bool if presencePayload, ok = payload.Payload.(map[string]interface{}); !ok { - m.listenerManager.announceStatus(&PNStatus{ - Category: PNUnknownCategory, - ErrorData: errors.New("Presence response parsing error"), - Error: true, - Operation: PNSubscribeOperation, - AffectedChannels: []string{channel}, - }) + announceSubscribePayloadParsingError(m, channel, "Presence response parsing error") + return } action, _ = presencePayload["action"].(string) @@ -579,20 +657,21 @@ func processPresencePayload(m *SubscriptionManager, payload subscribeMessage, ch switch presencePayload["timestamp"].(type) { case int: timestamp = int64(presencePayload["timestamp"].(int)) - break case int64: timestamp = presencePayload["timestamp"].(int64) - break case float64: timestamp = int64(presencePayload["timestamp"].(float64)) - break } } data = presencePayload["data"] if presencePayload["here_now_refresh"] != nil { - hereNowRefresh = presencePayload["here_now_refresh"].(bool) + hereNowRefresh, ok = presencePayload["here_now_refresh"].(bool) + if !ok { + announceSubscribePayloadParsingError(m, channel, "Presence response parsing error: invalid here_now_refresh") + return + } } timetoken, _ := strconv.ParseInt(publishMeta.PublishTimetoken, 10, 64) @@ -644,7 +723,10 @@ func processNonPresencePayload(m *SubscriptionManager, payload subscribeMessage, m.pubnub.loggerManager.LogSimple(PNLogLevelTrace, fmt.Sprintf("Announcing signal: channel=%s", channel), false) m.listenerManager.announceSignal(pnMessageResult) case PNMessageTypeObjects: - pnUUIDEvent, pnChannelEvent, pnMembershipEvent, eventType := createPNObjectsResult(payload.Payload, m, actualCh, subscribedCh, channel, subscriptionMatch) + pnUUIDEvent, pnChannelEvent, pnMembershipEvent, eventType, ok := createPNObjectsResult(payload.Payload, m, actualCh, subscribedCh, channel, subscriptionMatch) + if !ok { + return + } m.pubnub.loggerManager.LogSimple(PNLogLevelTrace, fmt.Sprintf("Announcing objects event: type=%v, channel=%s", eventType, channel), false) switch eventType { case PNObjectsUUIDEvent: @@ -658,7 +740,10 @@ func processNonPresencePayload(m *SubscriptionManager, payload subscribeMessage, m.listenerManager.announceMembershipEvent(pnMembershipEvent) } case PNMessageTypeMessageActions: - pnMessageActionsEvent := createPNMessageActionsEventResult(payload.Payload, m, actualCh, subscribedCh, channel, subscriptionMatch, payload.IssuingClientID) + pnMessageActionsEvent, ok := createPNMessageActionsEventResult(payload.Payload, m, actualCh, subscribedCh, channel, subscriptionMatch, payload.IssuingClientID) + if !ok { + return + } m.pubnub.loggerManager.LogSimple(PNLogLevelTrace, fmt.Sprintf("Announcing message actions event: channel=%s", channel), false) m.listenerManager.announceMessageActionsEvent(pnMessageActionsEvent) case PNMessageTypeFile: @@ -680,7 +765,10 @@ func processNonPresencePayload(m *SubscriptionManager, payload subscribeMessage, } - pnFilesEvent := createPNFilesEvent(messagePayload, m, actualCh, subscribedCh, channel, subscriptionMatch, payload.IssuingClientID, payload.UserMetadata, timetoken, err) + pnFilesEvent, ok := createPNFilesEvent(messagePayload, m, actualCh, subscribedCh, channel, subscriptionMatch, payload.IssuingClientID, payload.UserMetadata, timetoken, err) + if !ok { + return + } m.pubnub.loggerManager.LogSimple(PNLogLevelTrace, fmt.Sprintf("Announcing file event: channel=%s", channel), false) m.listenerManager.announceFile(pnFilesEvent) default: @@ -723,18 +811,12 @@ func processSubscribePayload(m *SubscriptionManager, payload subscribeMessage) { } } -func createPNFilesEvent(filePayload interface{}, m *SubscriptionManager, actualCh, subscribedCh, channel, subscriptionMatch, issuingClientID string, userMetadata interface{}, timetoken int64, err error) *PNFilesEvent { +func createPNFilesEvent(filePayload interface{}, m *SubscriptionManager, actualCh, subscribedCh, channel, subscriptionMatch, issuingClientID string, userMetadata interface{}, timetoken int64, err error) (*PNFilesEvent, bool) { var filesPayload map[string]interface{} var ok bool if filesPayload, ok = filePayload.(map[string]interface{}); !ok { - m.listenerManager.announceStatus(&PNStatus{ - Category: PNUnknownCategory, - ErrorData: errors.New("Files response parsing error"), - Error: true, - Operation: PNSubscribeOperation, - AffectedChannels: []string{channel}, - }) - return nil + announceSubscribePayloadParsingError(m, channel, "Files response parsing error") + return nil, false } resp := PNFileMessageAndDetails{} @@ -756,39 +838,58 @@ func createPNFilesEvent(filePayload interface{}, m *SubscriptionManager, actualC UserMetadata: userMetadata, Error: err, } - return pnFilesEvent + return pnFilesEvent, true } -func createPNMessageActionsEventResult(maPayload interface{}, m *SubscriptionManager, actualCh, subscribedCh, channel, subscriptionMatch, issuingClientID string) *PNMessageActionsEvent { +func createPNMessageActionsEventResult(maPayload interface{}, m *SubscriptionManager, actualCh, subscribedCh, channel, subscriptionMatch, issuingClientID string) (*PNMessageActionsEvent, bool) { var messageActionsPayload map[string]interface{} var ok bool if messageActionsPayload, ok = maPayload.(map[string]interface{}); !ok { - m.listenerManager.announceStatus(&PNStatus{ - Category: PNUnknownCategory, - ErrorData: errors.New("Message Actions response parsing error"), - Error: true, - Operation: PNSubscribeOperation, - AffectedChannels: []string{channel}, - }) - return nil + announceSubscribePayloadParsingError(m, channel, "Message Actions response parsing error") + return nil, false } - eventType := PNMessageActionsEventType(messageActionsPayload["event"].(string)) + event, ok := subscribePayloadRequiredString(messageActionsPayload, "event") + if !ok { + announceSubscribePayloadParsingError(m, channel, "Message Actions response parsing error: invalid event") + return nil, false + } + eventType := PNMessageActionsEventType(event) var data map[string]interface{} resp := PNMessageActionsResponse{} if o, ok := messageActionsPayload["data"]; ok { - data = o.(map[string]interface{}) + data, ok = subscribePayloadMap(o) + if !ok { + announceSubscribePayloadParsingError(m, channel, "Message Actions response parsing error: invalid data") + return nil, false + } if d, ok := data["type"]; ok { - resp.ActionType = d.(string) + resp.ActionType, ok = d.(string) + if !ok { + announceSubscribePayloadParsingError(m, channel, "Message Actions response parsing error: invalid type") + return nil, false + } } if d, ok := data["value"]; ok { - resp.ActionValue = d.(string) + resp.ActionValue, ok = d.(string) + if !ok { + announceSubscribePayloadParsingError(m, channel, "Message Actions response parsing error: invalid value") + return nil, false + } } if d, ok := data["actionTimetoken"]; ok { - resp.ActionTimetoken = d.(string) + resp.ActionTimetoken, ok = d.(string) + if !ok { + announceSubscribePayloadParsingError(m, channel, "Message Actions response parsing error: invalid actionTimetoken") + return nil, false + } } if d, ok := data["messageTimetoken"]; ok { - resp.MessageTimetoken = d.(string) + resp.MessageTimetoken, ok = d.(string) + if !ok { + announceSubscribePayloadParsingError(m, channel, "Message Actions response parsing error: invalid messageTimetoken") + return nil, false + } } resp.UUID = issuingClientID } @@ -802,82 +903,125 @@ func createPNMessageActionsEventResult(maPayload interface{}, m *SubscriptionMan Subscription: subscriptionMatch, } - return pnMessageActionsEvent + return pnMessageActionsEvent, true } -func createPNObjectsResult(objPayload interface{}, m *SubscriptionManager, actualCh, subscribedCh, channel, subscriptionMatch string) (*PNUUIDEvent, *PNChannelEvent, *PNMembershipEvent, PNObjectsEventType) { +func createPNObjectsResult(objPayload interface{}, m *SubscriptionManager, actualCh, subscribedCh, channel, subscriptionMatch string) (*PNUUIDEvent, *PNChannelEvent, *PNMembershipEvent, PNObjectsEventType, bool) { var objectsPayload map[string]interface{} var ok bool if objectsPayload, ok = objPayload.(map[string]interface{}); !ok { - m.listenerManager.announceStatus(&PNStatus{ - Category: PNUnknownCategory, - ErrorData: errors.New("Objects response parsing error"), - Error: true, - Operation: PNSubscribeOperation, - AffectedChannels: []string{channel}, - }) - return nil, nil, nil, PNObjectsNoneEvent + announceSubscribePayloadParsingError(m, channel, "Objects response parsing error") + return nil, nil, nil, PNObjectsNoneEvent, false + } + eventTypeString, ok := subscribePayloadRequiredString(objectsPayload, "type") + if !ok { + announceSubscribePayloadParsingError(m, channel, "Objects response parsing error: invalid type") + return nil, nil, nil, PNObjectsNoneEvent, false } - eventType := PNObjectsEventType(objectsPayload["type"].(string)) - event := PNObjectsEvent(objectsPayload["event"].(string)) + eventString, ok := subscribePayloadRequiredString(objectsPayload, "event") + if !ok { + announceSubscribePayloadParsingError(m, channel, "Objects response parsing error: invalid event") + return nil, nil, nil, PNObjectsNoneEvent, false + } + eventType := PNObjectsEventType(eventTypeString) + event := PNObjectsEvent(eventString) version := "" if d, ok := objectsPayload["version"]; ok { - version = d.(string) + version, ok = d.(string) + if !ok { + announceSubscribePayloadParsingError(m, channel, "Objects response parsing error: invalid version") + return nil, nil, nil, PNObjectsNoneEvent, false + } if version == "1.0" { m.pubnub.loggerManager.LogSimple(PNLogLevelDebug, "Ignoring objects event version 1.0", false) - return &PNUUIDEvent{}, &PNChannelEvent{}, &PNMembershipEvent{}, PNObjectsNoneEvent + return &PNUUIDEvent{}, &PNChannelEvent{}, &PNMembershipEvent{}, PNObjectsNoneEvent, true } } else { m.pubnub.loggerManager.LogSimple(PNLogLevelDebug, "Ignoring non-versioned objects event", false) - return &PNUUIDEvent{}, &PNChannelEvent{}, &PNMembershipEvent{}, PNObjectsNoneEvent + return &PNUUIDEvent{}, &PNChannelEvent{}, &PNMembershipEvent{}, PNObjectsNoneEvent, true } var id, UUID, channelID, description, timestamp, updated, eTag, name, externalID, profileURL, email, status, objectType string var custom, data map[string]interface{} if o, ok := objectsPayload["data"]; ok { - data = o.(map[string]interface{}) + data, ok = subscribePayloadMap(o) + if !ok { + announceSubscribePayloadParsingError(m, channel, "Objects response parsing error: invalid data") + return nil, nil, nil, PNObjectsNoneEvent, false + } if d, ok := data["uuid"]; ok { - u := d.(map[string]interface{}) - UUID = u["id"].(string) + u, ok := subscribePayloadMap(d) + if !ok { + announceSubscribePayloadParsingError(m, channel, "Objects response parsing error: invalid uuid") + return nil, nil, nil, PNObjectsNoneEvent, false + } + UUID, ok = subscribePayloadRequiredString(u, "id") + if !ok { + announceSubscribePayloadParsingError(m, channel, "Objects response parsing error: invalid uuid.id") + return nil, nil, nil, PNObjectsNoneEvent, false + } } if d, ok := data["id"]; ok { - id = d.(string) + id, ok = d.(string) + if !ok { + announceSubscribePayloadParsingError(m, channel, "Objects response parsing error: invalid id") + return nil, nil, nil, PNObjectsNoneEvent, false + } } if d, ok := data["channel"]; ok { - ch := d.(map[string]interface{}) - channelID = ch["id"].(string) + ch, ok := subscribePayloadMap(d) + if !ok { + announceSubscribePayloadParsingError(m, channel, "Objects response parsing error: invalid channel") + return nil, nil, nil, PNObjectsNoneEvent, false + } + channelID, ok = subscribePayloadRequiredString(ch, "id") + if !ok { + announceSubscribePayloadParsingError(m, channel, "Objects response parsing error: invalid channel.id") + return nil, nil, nil, PNObjectsNoneEvent, false + } } - if d, ok := data["name"]; ok { - name = d.(string) + if name, ok = subscribePayloadOptionalString(data, "name"); !ok { + announceSubscribePayloadParsingError(m, channel, "Objects response parsing error: invalid name") + return nil, nil, nil, PNObjectsNoneEvent, false } - if d, ok := data["externalId"]; ok { - externalID = d.(string) + if externalID, ok = subscribePayloadOptionalString(data, "externalId"); !ok { + announceSubscribePayloadParsingError(m, channel, "Objects response parsing error: invalid externalId") + return nil, nil, nil, PNObjectsNoneEvent, false } - if d, ok := data["profileUrl"]; ok { - profileURL = d.(string) + if profileURL, ok = subscribePayloadOptionalString(data, "profileUrl"); !ok { + announceSubscribePayloadParsingError(m, channel, "Objects response parsing error: invalid profileUrl") + return nil, nil, nil, PNObjectsNoneEvent, false } - if d, ok := data["email"]; ok { - email = d.(string) + if email, ok = subscribePayloadOptionalString(data, "email"); !ok { + announceSubscribePayloadParsingError(m, channel, "Objects response parsing error: invalid email") + return nil, nil, nil, PNObjectsNoneEvent, false } - if d, ok := data["description"]; ok { - description = d.(string) + if description, ok = subscribePayloadOptionalString(data, "description"); !ok { + announceSubscribePayloadParsingError(m, channel, "Objects response parsing error: invalid description") + return nil, nil, nil, PNObjectsNoneEvent, false } - if d, ok := data["timestamp"]; ok { - timestamp = d.(string) + if timestamp, ok = subscribePayloadOptionalString(data, "timestamp"); !ok { + announceSubscribePayloadParsingError(m, channel, "Objects response parsing error: invalid timestamp") + return nil, nil, nil, PNObjectsNoneEvent, false } - if d, ok := data["updated"]; ok { - updated = d.(string) + if updated, ok = subscribePayloadOptionalString(data, "updated"); !ok { + announceSubscribePayloadParsingError(m, channel, "Objects response parsing error: invalid updated") + return nil, nil, nil, PNObjectsNoneEvent, false } - if d, ok := data["eTag"]; ok { - eTag = d.(string) + if eTag, ok = subscribePayloadOptionalString(data, "eTag"); !ok { + announceSubscribePayloadParsingError(m, channel, "Objects response parsing error: invalid eTag") + return nil, nil, nil, PNObjectsNoneEvent, false } - if d, ok := data["custom"]; ok { - custom = d.(map[string]interface{}) + if custom, ok = subscribePayloadOptionalMap(data, "custom"); !ok { + announceSubscribePayloadParsingError(m, channel, "Objects response parsing error: invalid custom") + return nil, nil, nil, PNObjectsNoneEvent, false } - if d, ok := data["status"]; ok { - status = d.(string) + if status, ok = subscribePayloadOptionalString(data, "status"); !ok { + announceSubscribePayloadParsingError(m, channel, "Objects response parsing error: invalid status") + return nil, nil, nil, PNObjectsNoneEvent, false } - if d, ok := data["type"]; ok { - objectType = d.(string) + if objectType, ok = subscribePayloadOptionalString(data, "type"); !ok { + announceSubscribePayloadParsingError(m, channel, "Objects response parsing error: invalid object type") + return nil, nil, nil, PNObjectsNoneEvent, false } } @@ -952,7 +1096,7 @@ func createPNObjectsResult(objPayload interface{}, m *SubscriptionManager, actua Subscription: subscriptionMatch, } - return pnUUIDEvent, pnChannelEvent, pnMembershipEvent, eventType + return pnUUIDEvent, pnChannelEvent, pnMembershipEvent, eventType, true } func createPNMessageResult(messagePayload interface{}, actualCh, subscribedCh, channel, subscriptionMatch, issuingClientID string, userMetadata interface{}, timetoken int64, CustomMessageType string, error error) *PNMessage { diff --git a/subscription_manager_test.go b/subscription_manager_test.go index 56527027..4dd7bef7 100644 --- a/subscription_manager_test.go +++ b/subscription_manager_test.go @@ -5,6 +5,7 @@ import ( "log" "reflect" "testing" + "time" "github.com/pubnub/go/v9/crypto" "github.com/stretchr/testify/assert" @@ -608,6 +609,320 @@ func TestProcessSubscribePayloadWithCustomMessageType(t *testing.T) { <-done } +func waitForSubscribeStatus(t *testing.T, listener *Listener) *PNStatus { + t.Helper() + + select { + case status := <-listener.Status: + return status + case <-time.After(time.Second): + t.Fatal("timed out waiting for subscribe status") + return nil + } +} + +func assertUnknownSubscribeStatus(t *testing.T, listener *Listener, channel string) { + t.Helper() + + status := waitForSubscribeStatus(t, listener) + assert.True(t, status.Error) + assert.Equal(t, PNUnknownCategory, status.Category) + assert.Equal(t, PNSubscribeOperation, status.Operation) + assert.Equal(t, []string{channel}, status.AffectedChannels) +} + +func TestProcessPresencePayloadInvalidHereNowRefreshAnnouncesUnknownStatus(t *testing.T) { + listener := NewListener() + pn := NewPubNub(NewDemoConfig()) + pn.AddListener(listener) + + sm := &subscribeMessage{ + Shard: "1", + SubscriptionMatch: "cg-pnpres", + Channel: "channel-pnpres", + Payload: map[string]interface{}{ + "action": "join", + "timestamp": int64(15078947309567840), + "uuid": "bfce00ff4018fce180438bb04afc8da8", + "occupancy": float64(1), + "here_now_refresh": "true", + }, + } + + processSubscribePayload(pn.subscriptionManager, *sm) + + assertUnknownSubscribeStatus(t, listener, "channel-pnpres") + select { + case presence := <-listener.Presence: + assert.Failf(t, "unexpected presence event", "event: %#v", presence) + case <-time.After(50 * time.Millisecond): + } +} + +func TestProcessMessageActionsPayloadMalformedFieldsAnnounceUnknownStatus(t *testing.T) { + tests := []struct { + name string + payload interface{} + }{ + { + name: "invalid event", + payload: map[string]interface{}{ + "event": 42, + }, + }, + { + name: "invalid data", + payload: map[string]interface{}{ + "event": "added", + "data": nil, + }, + }, + { + name: "invalid action type", + payload: map[string]interface{}{ + "event": "added", + "data": map[string]interface{}{ + "type": nil, + }, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + listener := NewListener() + pn := NewPubNub(NewDemoConfig()) + pn.AddListener(listener) + + sm := &subscribeMessage{ + Shard: "1", + SubscriptionMatch: "channel", + Channel: "channel", + Payload: tt.payload, + MessageType: PNMessageTypeMessageActions, + } + + processSubscribePayload(pn.subscriptionManager, *sm) + + assertUnknownSubscribeStatus(t, listener, "channel") + select { + case event := <-listener.MessageActionsEvent: + assert.Failf(t, "unexpected message actions event", "event: %#v", event) + case <-time.After(50 * time.Millisecond): + } + }) + } +} + +func TestProcessMessageActionsPayloadValidPayloadAnnouncesEvent(t *testing.T) { + listener := NewListener() + pn := NewPubNub(NewDemoConfig()) + pn.AddListener(listener) + + sm := &subscribeMessage{ + Shard: "1", + SubscriptionMatch: "channel", + Channel: "channel", + IssuingClientID: "uuid-1", + Payload: map[string]interface{}{ + "event": "added", + "data": map[string]interface{}{ + "type": "reaction", + "value": "smile", + "actionTimetoken": "123", + "messageTimetoken": "456", + }, + }, + MessageType: PNMessageTypeMessageActions, + } + + processSubscribePayload(pn.subscriptionManager, *sm) + + select { + case event := <-listener.MessageActionsEvent: + assert.Equal(t, PNMessageActionsAdded, event.Event) + assert.Equal(t, "reaction", event.Data.ActionType) + assert.Equal(t, "smile", event.Data.ActionValue) + assert.Equal(t, "123", event.Data.ActionTimetoken) + assert.Equal(t, "456", event.Data.MessageTimetoken) + assert.Equal(t, "uuid-1", event.Data.UUID) + case <-time.After(time.Second): + t.Fatal("timed out waiting for message actions event") + } + + select { + case status := <-listener.Status: + assert.Failf(t, "unexpected status", "status: %#v", status) + case <-time.After(50 * time.Millisecond): + } +} + +func TestProcessObjectsPayloadMalformedFieldsAnnounceUnknownStatus(t *testing.T) { + tests := []struct { + name string + payload interface{} + }{ + { + name: "invalid type", + payload: map[string]interface{}{ + "type": 42, + "event": "set", + "version": "2.0", + }, + }, + { + name: "invalid data", + payload: map[string]interface{}{ + "type": "uuid", + "event": "set", + "version": "2.0", + "data": nil, + }, + }, + { + name: "invalid nested uuid", + payload: map[string]interface{}{ + "type": "membership", + "event": "set", + "version": "2.0", + "data": map[string]interface{}{ + "uuid": nil, + }, + }, + }, + { + name: "invalid custom", + payload: map[string]interface{}{ + "type": "uuid", + "event": "set", + "version": "2.0", + "data": map[string]interface{}{ + "id": "uuid-1", + "custom": "not-a-map", + }, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + listener := NewListener() + pn := NewPubNub(NewDemoConfig()) + pn.AddListener(listener) + + sm := &subscribeMessage{ + Shard: "1", + SubscriptionMatch: "channel", + Channel: "channel", + Payload: tt.payload, + MessageType: PNMessageTypeObjects, + } + + processSubscribePayload(pn.subscriptionManager, *sm) + + assertUnknownSubscribeStatus(t, listener, "channel") + select { + case event := <-listener.UUIDEvent: + assert.Failf(t, "unexpected uuid event", "event: %#v", event) + case event := <-listener.ChannelEvent: + assert.Failf(t, "unexpected channel event", "event: %#v", event) + case event := <-listener.MembershipEvent: + assert.Failf(t, "unexpected membership event", "event: %#v", event) + case <-time.After(50 * time.Millisecond): + } + }) + } +} + +func TestProcessObjectsPayloadValidUUIDPayloadAnnouncesEvent(t *testing.T) { + listener := NewListener() + pn := NewPubNub(NewDemoConfig()) + pn.AddListener(listener) + + sm := &subscribeMessage{ + Shard: "1", + SubscriptionMatch: "channel", + Channel: "channel", + Payload: map[string]interface{}{ + "type": "uuid", + "event": "set", + "version": "2.0", + "data": map[string]interface{}{ + "id": "uuid-1", + "name": "User One", + "custom": map[string]interface{}{ + "role": "admin", + }, + }, + }, + MessageType: PNMessageTypeObjects, + } + + processSubscribePayload(pn.subscriptionManager, *sm) + + select { + case event := <-listener.UUIDEvent: + assert.Equal(t, PNObjectsEvent(PNObjectsEventSet), event.Event) + assert.Equal(t, "uuid-1", event.UUID) + assert.Equal(t, "User One", event.Name) + assert.Equal(t, "admin", event.Custom["role"]) + assert.Equal(t, "channel", event.Channel) + case <-time.After(time.Second): + t.Fatal("timed out waiting for uuid event") + } + + select { + case status := <-listener.Status: + assert.Failf(t, "unexpected status", "status: %#v", status) + case <-time.After(50 * time.Millisecond): + } +} + +func TestProcessFilePayloadMalformedPayloadAnnouncesUnknownStatus(t *testing.T) { + listener := NewListener() + pn := NewPubNub(NewDemoConfig()) + pn.AddListener(listener) + + sm := &subscribeMessage{ + Shard: "1", + SubscriptionMatch: "channel", + Channel: "channel", + Payload: "not-a-file-payload", + MessageType: PNMessageTypeFile, + } + + processSubscribePayload(pn.subscriptionManager, *sm) + + assertUnknownSubscribeStatus(t, listener, "channel") + select { + case event := <-listener.File: + assert.Failf(t, "unexpected file event", "event: %#v", event) + case <-time.After(50 * time.Millisecond): + } +} + +func TestSafeProcessSubscribePayloadRecoversAndAnnouncesUnknownStatus(t *testing.T) { + listener := NewListener() + pn := NewPubNub(NewDemoConfig()) + cryptoModule, err := crypto.NewAesCbcCryptoModule("enigma", true) + assert.Nil(t, err) + pn.Config.CryptoModule = cryptoModule + pn.AddListener(listener) + + sm := &subscribeMessage{ + Shard: "1", + SubscriptionMatch: "channel", + Channel: "channel", + Payload: nil, + } + + assert.NotPanics(t, func() { + safeProcessSubscribePayload(pn.subscriptionManager, *sm) + }) + + assertUnknownSubscribeStatus(t, listener, "channel") +} + func TestDecryptionProcessOnEncryptedMessage(t *testing.T) { assert := assert.New(t) pn := NewPubNub(NewDemoConfig()) diff --git a/utils/string_utils.go b/utils/string_utils.go index e01fa209..a13ae09c 100644 --- a/utils/string_utils.go +++ b/utils/string_utils.go @@ -102,9 +102,23 @@ func UUID() string { return fmt.Sprintf("pn-%s", uuid.New().String()) } -func CheckUUID(uuid string) { +// ValidateUUID reports whether the supplied UUID is usable. It returns a +// BuildRequestError when the UUID is empty or whitespace-only and nil otherwise. +func ValidateUUID(uuid string) error { if strings.TrimSpace(uuid) == "" { - panic(pnerr.NewBuildRequestError("uuid: UUID misisng, please set it in the PNConfig.")) + return pnerr.NewBuildRequestError("uuid: UUID missing, please set it in the PNConfig.") + } + return nil +} + +// CheckUUID panics when the supplied UUID is empty or whitespace-only. +// +// Deprecated: CheckUUID panics on invalid input, which can crash the process +// when reached from an SDK-owned goroutine (e.g. the subscribe loop). Use +// ValidateUUID and propagate the returned error instead. +func CheckUUID(uuid string) { + if err := ValidateUUID(uuid); err != nil { + panic(err) } }