fix(rtdb): percent-encode special characters in query values

firebase_admin/db.py

@@ -23,6 +23,7 @@
 import collections
 import json
 import os
+import string
 import sys
 import threading
 from urllib import parse
@@ -71,6 +72,9 @@ def reference(path='/', app=None, url=None):
     client = service.get_client(url)
     return Reference(client=client, path=path)
 
+_QUERY_VALUE_SAFE_CHARS = ''.join(c for c in string.punctuation if c not in '#%&+=?')
+
+
 def _parse_path(path):
     """Parses a path string into a set of segments."""
     if not isinstance(path, str):
@@ -603,7 +607,10 @@ def equal_to(self, value):
     def _querystr(self):
         params = []
         for key in sorted(self._params):
-            params.append(f'{key}={self._params[key]}')
+            # Percent-encode the characters that have a special meaning in a query string
+            # (e.g. '&', '#' and '+'), so that they are sent as part of the value.
+            value = parse.quote(str(self._params[key]), safe=_QUERY_VALUE_SAFE_CHARS)
+            params.append(f'{key}={value}')
         return '&'.join(params)
 
     def get(self):

tests/test_db.py

@@ -18,6 +18,7 @@
 import os
 import sys
 import time
+from urllib import parse
 
 import pytest
 
@@ -993,6 +994,20 @@ def test_valid_start_at(self, arg):
         query = self.ref.order_by_child('foo').start_at(arg)
         assert query._querystr == f'orderBy="foo"&startAt={json.dumps(arg)}'
 
+    @pytest.mark.parametrize('value, encoded', [
+        ('C#', '"C%23"'),
+        ('+15555555', '"%2B15555555"'),
+        ('a&b=c', '"a%26b%3Dc"'),
+        ('100%', '"100%25"'),
+        ('what?', '"what%3F"'),
+        ('a b', '"a%20b"'),
+    ])
+    def test_special_characters_are_encoded(self, value, encoded):
+        query = self.ref.order_by_child('foo').equal_to(value)
+        assert query._querystr == f'equalTo={encoded}&orderBy="foo"'
+        parsed = parse.parse_qs(query._querystr)
+        assert json.loads(parsed['equalTo'][0]) == value
+
     def test_end_at_none(self):
         query = self.ref.order_by_child('foo')
         with pytest.raises(ValueError):