Fix convert processor conversion of string with leading zeros to inte…

…ger (elastic#15557)

The conversion failed when for strings with leading zeroes and a decimal
digit 8 or 9, as the underlying runtime function would try to parse that
as an octal number.

This is fixed by only allowing decimal and hex, which in turns makes the
processor more aligned to its Elasticsearch counterpart.

Fixes elastic#15513

(cherry picked from commit e0071b5)

CHANGELOG.next.asciidoc

@@ -44,6 +44,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 
 - Fix a race condition with the Kafka pipeline client, it is possible that `Close()` get called before `Connect()` . {issue}11945[11945]
 - Allow users to configure only `cluster_uuid` setting under `monitoring` namespace. {pull}14338[14338]
+- Fix `convert` processor conversion of string to integer with leading zeros. {issue}15513[15513] {pull}15557[15557]
 
 *Auditbeat*
 

libbeat/processors/convert/convert.go

@@ -205,7 +205,7 @@ func toString(value interface{}) (string, error) {
 func toLong(value interface{}) (int64, error) {
 	switch v := value.(type) {
 	case string:
-		return strconv.ParseInt(v, 0, 64)
+		return strToInt(v, 64)
 	case int:
 		return int64(v), nil
 	case int8:
@@ -238,7 +238,7 @@ func toLong(value interface{}) (int64, error) {
 func toInteger(value interface{}) (int32, error) {
 	switch v := value.(type) {
 	case string:
-		i, err := strconv.ParseInt(v, 0, 32)
+		i, err := strToInt(v, 32)
 		return int32(i), err
 	case int:
 		return int32(v), nil
@@ -403,3 +403,24 @@ func cloneValue(value interface{}) interface{} {
 		return value
 	}
 }
+
+// strToInt is a helper to interpret a string as either base 10 or base 16.
+func strToInt(s string, bitSize int) (int64, error) {
+	base := 10
+	if hasHexPrefix(s) {
+		// strconv.ParseInt will accept the '0x' or '0X` prefix only when base is 0.
+		base = 0
+	}
+	return strconv.ParseInt(s, base, bitSize)
+}
+
+func hasHexPrefix(s string) bool {
+	if len(s) < 3 {
+		return false
+	}
+	a, b := s[0], s[1]
+	if a == '+' || a == '-' {
+		a, b = b, s[2]
+	}
+	return a == '0' && (b == 'x' || b == 'X')
+}

libbeat/processors/convert/convert_test.go

@@ -276,8 +276,16 @@ var testCases = []testCase{
 
 	{Long, nil, nil, true},
 	{Long, "x", nil, true},
+	{Long, "0x", nil, true},
+	{Long, "0b1", nil, true},
+	{Long, "1x2", nil, true},
 	{Long, true, nil, true},
 	{Long, "1", int64(1), false},
+	{Long, "-1", int64(-1), false},
+	{Long, "017", int64(17), false},
+	{Long, "08", int64(8), false},
+	{Long, "0X0A", int64(10), false},
+	{Long, "-0x12", int64(-18), false},
 	{Long, int(1), int64(1), false},
 	{Long, int8(1), int64(1), false},
 	{Long, int16(1), int64(1), false},
@@ -294,6 +302,17 @@ var testCases = []testCase{
 	{Integer, nil, nil, true},
 	{Integer, "x", nil, true},
 	{Integer, true, nil, true},
+	{Integer, "x", nil, true},
+	{Integer, "0x", nil, true},
+	{Integer, "0b1", nil, true},
+	{Integer, "1x2", nil, true},
+	{Integer, true, nil, true},
+	{Integer, "1", int32(1), false},
+	{Integer, "-1", int32(-1), false},
+	{Integer, "017", int32(17), false},
+	{Integer, "08", int32(8), false},
+	{Integer, "0X0A", int32(10), false},
+	{Integer, "-0x12", int32(-18), false},
 	{Integer, "1", int32(1), false},
 	{Integer, int(1), int32(1), false},
 	{Integer, int8(1), int32(1), false},