Fix unescaping (prometheus#291)

immerrr · brian-brazil · commit f2436cacdefe · 2018-07-24T15:14:08.000+01:00
* Extend parser tests
* parser: fix handing of escape sequences

Signed-off-by: immerrr &lt;immerrr@gmail.com&gt;
diff --git a/prometheus_client/parser.py b/prometheus_client/parser.py
@@ -2,6 +2,8 @@
 
 from __future__ import unicode_literals
 
+import re
+
 try:
     import StringIO
 except ImportError:
@@ -19,11 +21,37 @@ def text_string_to_metric_families(text):
     for metric_family in text_fd_to_metric_families(StringIO.StringIO(text)):
         yield metric_family
 
+
+ESCAPE_SEQUENCES = {
+    '\\\\': '\\',
+    '\\n': '\n',
+    '\\"': '"',
+}
+
+
+def replace_escape_sequence(match):
+    return ESCAPE_SEQUENCES[match.group(0)]
+
+
+HELP_ESCAPING_RE = re.compile(r'\\[\\n]')
+ESCAPING_RE = re.compile(r'\\[\\n"]')
+
+
 def _replace_help_escaping(s):
-    return s.replace("\\n", "\n").replace('\\\\', '\\')
+    return HELP_ESCAPING_RE.sub(replace_escape_sequence, s)
+
 
 def _replace_escaping(s):
-    return s.replace("\\n", "\n").replace('\\\\', '\\').replace('\\"', '"')
+    return ESCAPING_RE.sub(replace_escape_sequence, s)
+
+
+def _is_character_escaped(s, charpos):
+    num_bslashes = 0
+    while (charpos > num_bslashes and
+           s[charpos - 1 - num_bslashes] == '\\'):
+        num_bslashes += 1
+    return num_bslashes % 2 == 1
+
 
 def _parse_labels(labels_string):
     labels = {}
@@ -52,7 +80,7 @@ def _parse_labels(labels_string):
             i = 0
             while i < len(value_substr):
                 i = value_substr.index('"', i)
-                if value_substr[i - 1] != "\\":
+                if not _is_character_escaped(value_substr, i):
                     break
                 i += 1
 
@@ -62,7 +90,7 @@ def _parse_labels(labels_string):
             # Replace escaping if needed
             if escaping:
                 label_value = _replace_escaping(label_value)
-            labels[label_name.strip()] = label_value.strip()
+            labels[label_name.strip()] = label_value
 
             # Remove the processed label from the sub-slice for next iteration
             sub_labels = sub_labels[quote_end + 1:]
diff --git a/tests/test_parser.py b/tests/test_parser.py
@@ -167,14 +167,32 @@ def test_commas(self):
         families = text_string_to_metric_families("""# TYPE a counter
 # HELP a help
 a{foo="bar",} 1
+a{foo="baz",  } 1
 # TYPE b counter
 # HELP b help
 b{,} 2
+# TYPE c counter
+# HELP c help
+c{  ,} 3
+# TYPE d counter
+# HELP d help
+d{,  } 4
 """)
         a = CounterMetricFamily("a", "help", labels=["foo"])
         a.add_metric(["bar"], 1)
+        a.add_metric(["baz"], 1)
         b = CounterMetricFamily("b", "help", value=2)
-        self.assertEqual([a, b], list(families))
+        c = CounterMetricFamily("c", "help", value=3)
+        d = CounterMetricFamily("d", "help", value=4)
+        self.assertEqual([a, b, c, d], list(families))
+
+    def test_multiple_trailing_commas(self):
+        text = """# TYPE a counter
+# HELP a help
+a{foo="bar",, } 1
+"""
+        self.assertRaises(ValueError,
+                          lambda: list(text_string_to_metric_families(text)))
 
     def test_empty_brackets(self):
         families = text_string_to_metric_families("""# TYPE a counter
@@ -200,6 +218,50 @@ def test_empty_label(self):
         metric_family.add_metric([""], 2)
         self.assertEqual([metric_family], list(families))
 
+    def test_label_escaping(self):
+        for escaped_val, unescaped_val in [
+                ('foo', 'foo'),
+                ('\\foo', '\\foo'),
+                ('\\\\foo', '\\foo'),
+                ('foo\\\\', 'foo\\'),
+                ('\\\\', '\\'),
+                ('\\n', '\n'),
+                ('\\\\n', '\\n'),
+                ('\\\\\\n', '\\\n'),
+                ('\\"', '"'),
+                ('\\\\\\"', '\\"')]:
+            families = list(text_string_to_metric_families("""
+# TYPE a counter
+# HELP a help
+a{foo="%s",bar="baz"} 1
+""" % escaped_val))
+            metric_family = CounterMetricFamily(
+                "a", "help", labels=["foo", "bar"])
+            metric_family.add_metric([unescaped_val, "baz"], 1)
+            self.assertEqual([metric_family], list(families))
+
+    def test_help_escaping(self):
+        for escaped_val, unescaped_val in [
+                ('foo', 'foo'),
+                ('\\foo', '\\foo'),
+                ('\\\\foo', '\\foo'),
+                ('foo\\', 'foo\\'),
+                ('foo\\\\', 'foo\\'),
+                ('\\n', '\n'),
+                ('\\\\n', '\\n'),
+                ('\\\\\\n', '\\\n'),
+                ('\\"', '\\"'),
+                ('\\\\"', '\\"'),
+                ('\\\\\\"', '\\\\"')]:
+            families = list(text_string_to_metric_families("""
+# TYPE a counter
+# HELP a %s
+a{foo="bar"} 1
+""" % escaped_val))
+            metric_family = CounterMetricFamily("a", unescaped_val, labels=["foo"])
+            metric_family.add_metric(["bar"], 1)
+            self.assertEqual([metric_family], list(families))
+
     def test_escaping(self):
         families = text_string_to_metric_families("""# TYPE a counter
 # HELP a he\\n\\\\l\\tp
