Allow only http and https in the callback URL, and test the zone and port rules

pipeline/fiftyone_pipeline_core/src/fiftyone_pipeline_core/callback_url.py

@@ -24,9 +24,10 @@
 The callback URL the client script posts to for updated JSON. The protocol
 and host usually come from the request, and the script holds the URL in a
 single-quoted string, so the URL is checked before it is written into the
-script. A URL that is not valid is refused. In the user information, in
-anything after the host and port, and in the endpoint, the characters a URL
-cannot hold or that could end the string or the script are percent-encoded.
+script. A URL that is not valid, including one whose protocol is not http or
+https, is refused. In the user information, in anything after the host and
+port, and in the endpoint, the characters a URL cannot hold or that could end
+the string or the script are percent-encoded.
 
 See https://github.com/51Degrees/specifications/blob/main/pipeline-specification/pipeline-elements/javascript-builder.md#callback-url
 """
@@ -58,9 +59,13 @@
     r"[\t-\r \u0085\u00a0\u1680\u2000-\u200a"
     r"\u2028\u2029\u202f\u205f\u3000]*")
 
-# A scheme as RFC 3986 section 3.1 defines one, being a letter followed by
-# letters, digits, '+', '-' and '.'.
-SCHEME_PATTERN = re.compile(r"[A-Za-z][A-Za-z0-9+.-]*")
+# The schemes the script can send its request with. It uses the URL only with
+# fetch and XMLHttpRequest, which reach a server with no other scheme.
+SCHEMES = ("http", "https")
+
+# Lower-cases the ASCII letters A to Z and nothing else.
+_ASCII_LOWER = str.maketrans(
+    "ABCDEFGHIJKLMNOPQRSTUVWXYZ", "abcdefghijklmnopqrstuvwxyz")
 
 # Where the authority ends, which is at the first '/', '?' or '#' of the
 # host.
@@ -105,6 +110,37 @@ def is_blank(text):
             or BLANK_PATTERN.fullmatch(text) is not None)
 
 
+def ascii_lower(text):
+
+    """!
+    The text with the ASCII letters A to Z in lower case and every other
+    character as it is, so that no letter outside ASCII can stand in for an
+    ASCII one, as str.casefold would let the long s stand in for "s".
+
+    @type text: str
+    @param text: The text
+    @rtype: str
+    @return: The text with A to Z in lower case
+    """
+
+    return text.translate(_ASCII_LOWER)
+
+
+def is_scheme(protocol):
+
+    """!
+    Whether the protocol is http or https, ignoring the case of its ASCII
+    letters only.
+
+    @type protocol: str
+    @param protocol: The protocol
+    @rtype: bool
+    @return: True if the protocol is http or https
+    """
+
+    return ascii_lower(protocol) in SCHEMES
+
+
 def callback_url(protocol, host, endpoint):
 
     """!
@@ -114,7 +150,7 @@ def callback_url(protocol, host, endpoint):
     the rest as given, apart from the percent-encoding.
 
     @type protocol: str
-    @param protocol: The protocol, such as https
+    @param protocol: The protocol, http or https in any case
     @type host: str
     @param host: The host, which can start with user information and end
     with a port, a path, a query or a fragment
@@ -134,7 +170,7 @@ def callback_url(protocol, host, endpoint):
     elif host_has_slash and endpoint_has_slash:
         endpoint = endpoint[1:]
     built = protocol + "://" + host + endpoint
-    if SCHEME_PATTERN.fullmatch(protocol) is None:
+    if not is_scheme(protocol):
         return CallbackUrl(False, built)
     match = AUTHORITY_END_PATTERN.search(host)
     end = len(host) if match is None else match.start()
@@ -143,7 +179,7 @@ def callback_url(protocol, host, endpoint):
         return CallbackUrl(False, built)
     return CallbackUrl(
         True,
-        protocol.lower() + "://" + authority
+        ascii_lower(protocol) + "://" + authority
         + _encoded(host[end:]) + _encoded(endpoint))
 
 

pipeline/fiftyone_pipeline_core/src/fiftyone_pipeline_core/javascriptbuilder.py

@@ -38,7 +38,7 @@
 from .evidence_keyfilter import EvidenceKeyFilter
 from .elementdata_dictionary import ElementDataDictionary
 from .constants import Constants
-from .callback_url import callback_url
+from .callback_url import callback_url, is_scheme
 
 
 # Evidence keys the rendered script is not configured with. Both are
@@ -255,9 +255,10 @@ def __init__(self, settings = {} ):
         * starting with a digit) and not a reserved word. An invalid
         * configured name raises a ValueError, and an invalid requested name
         * is ignored with a warning logged.
-        * @param {string} options.protocol The protocol ("http" or "https")
-        * used by the client side callback url.
-        * This can be overriden with header.protocol evidence
+        * @param {string} options.protocol The protocol of the client side
+        * callback url, http or https in any case of its ASCII letters. Any
+        * other configured protocol raises a ValueError. Left out, the
+        * header.protocol evidence is used, or https when there is none.
         * @param {string} options.host The host of the client side
         * callback url. This can be overriden with header.host evidence.
         * @param {string} options.endpoint The endpoint of the client side
@@ -286,6 +287,15 @@ def __init__(self, settings = {} ):
                 "JavaScript identifier (ASCII letters, digits, '_' and '$', "
                 "not starting with a digit) and not a reserved word.")
         self.settings['_protocol'] = settings["protocol"] if "protocol" in settings else None
+
+        # A protocol given here is used for every request, so one the script
+        # cannot send its request with is refused now rather than logged on
+        # the first request.
+        protocol = self.settings['_protocol']
+        if protocol and not (isinstance(protocol, str) and is_scheme(protocol)):
+            raise ValueError(
+                "The JavaScript builder's protocol setting must be http or "
+                "https.")
         self.settings['_host'] = settings["host"] if "host" in settings else None
         self.settings['_endpoint'] = settings["endpoint"] if "endpoint" in settings else ''
         self.settings['_enableCookies'] = settings["enable_cookies"] if "enable_cookies" in settings else True

pipeline/fiftyone_pipeline_core/tests/test_callback_url.py

@@ -137,19 +137,90 @@ def test_accepts_the_host(self, host, expected):
     def test_refuses_the_host(self, host):
         self.assertFalse(_url("https", host).valid, repr(host))
 
+    @parameterized.expand([
+        ("[fe80::1%25]",),
+        ("[fe80::1%25'+alert(1)+']",),
+        ("[fe80::1%eth0]",),
+        ("[fe80::1%25eth%0]",),
+        ("[::1]x",),
+        ("[::1]80",),
+        ("[::1]:80x",),
+        ("[::1]]",),
+    ])
+    def test_refuses_the_zone_or_the_text_after_the_bracket(self, host):
+        """Inside the brackets only an IPv6 address, optionally followed by
+        '%25' and a zone of unreserved characters, and after them only a
+        port. The zone check is what keeps a quote out of the script for a
+        bracketed host."""
+
+        self.assertFalse(_url("https", host).valid, repr(host))
+
+    def test_accepts_a_zone_followed_by_a_port(self):
+        self.assertEqual(
+            _valid("https://[fe80::1%25eth0]:8080/51dpipeline/json"),
+            _url("https", "[fe80::1%25eth0]:8080"))
+
+    def test_checks_the_port_by_its_value(self):
+        """A port goes by its value, so leading zeros are allowed, and a
+        port too long to read is refused rather than read as small enough."""
+
+        self.assertEqual(
+            _valid("https://example.com:00080/51dpipeline/json"),
+            _url("https", "example.com:00080"))
+        self.assertEqual(
+            _valid("https://example.com:0065535/51dpipeline/json"),
+            _url("https", "example.com:0065535"))
+        for host in [
+                "example.com:0065536",
+                "example.com:123456",
+                "example.com:4294967296",
+                "example.com:99999999999999999999"]:
+            self.assertFalse(_url("https", host).valid, host)
+
+    def test_encodes_the_characters_rule_8_lists(self):
+        """Each character rule 8 lists, in the user information, after the
+        host and in the endpoint, and a '%' without two hexadecimal digits
+        after it, even at the end."""
+
+        self.assertEqual(
+            _valid("https://u%22%5E%60%7B%7C%7D@example.com/%22%5E%60%7B%7C%7D"
+                   "/j%22%5E%60%7B%7C%7D"),
+            callback_url(
+                "https", 'u"^`{|}@example.com/"^`{|}', '/j"^`{|}'))
+        self.assertEqual(
+            _valid("https://example.com/a%25/51dpipeline/json"),
+            _url("https", "example.com/a%"))
+        self.assertEqual(
+            _valid("https://example.com/a%254/51dpipeline/json"),
+            _url("https", "example.com/a%4"))
+        self.assertEqual(
+            _valid("https://example.com/a%41/51dpipeline/json"),
+            _url("https", "example.com/a%41"))
+        self.assertEqual(
+            _valid("https://example.com/json%25"),
+            callback_url("https", "example.com", "/json%"))
+
     def test_the_invalid_url_is_the_one_built(self):
         self.assertEqual(
             CallbackUrl(False, "https://exa mple.com/51dpipeline/json"),
             _url("https", "exa mple.com"))
 
     def test_checks_the_scheme(self):
-        for protocol in ["https", "http", "ftp", "h+t-p.s"]:
+        """Only http and https, ignoring the case of ASCII letters only, so
+        a letter outside ASCII that a case-insensitive comparison can match
+        to an ASCII one, such as the long s, is refused."""
+
+        for protocol in ["https", "http", "HtTp"]:
             self.assertTrue(_url(protocol, "example.com").valid, protocol)
         self.assertEqual(
             _valid("https://example.com/51dpipeline/json"),
             _url("HTTPS", "example.com"))
-        for protocol in ["1http", "h t", "x');alert(1);//", "a://b"]:
-            self.assertFalse(_url(protocol, "example.com").valid, protocol)
+        for protocol in [
+                "ftp", "h+t-p.s", "ws", "javascript", "httpx", "https ",
+                "1http", "h t", "x');alert(1);//", "a://b", "http\u017f",
+                "HTTP\u017f"]:
+            self.assertFalse(
+                _url(protocol, "example.com").valid, repr(protocol))
 
     @parameterized.expand([
         ("example.com/');alert(1);//",),

pipeline/fiftyone_pipeline_core/tests/test_javascriptbuilder_variables.py

@@ -24,6 +24,8 @@
 import unittest
 from unittest import mock
 
+from parameterized import parameterized
+
 from fiftyone_pipeline_core import javascriptbuilder
 from fiftyone_pipeline_core.pipelinebuilder import PipelineBuilder
 
@@ -211,6 +213,33 @@ def test_refused_url_is_logged_on_one_line(self):
             self.assertNotIn(c, errors[0])
         self.assertIn("a\\nb.com\\r\\u2028", errors[0])
 
+    def test_refused_protocol_from_the_request_is_left_out_and_logged_once(
+            self):
+        """A protocol from the request that is neither http nor https gives
+        a script with no callback, and is logged once like any other URL
+        that is not valid."""
+
+        logger = MemoryLogger("error")
+        pipeline = PipelineBuilder({"javascript_builder_settings": {
+            "minify": False, "endpoint": "/json"}}).add_logger(logger).build()
+        captured = _CapturedAll()
+        with mock.patch.object(
+                javascriptbuilder.chevron, "render", captured.render):
+            for _ in range(2):
+                flowdata = pipeline.create_flowdata()
+                flowdata.evidence.add("header.host", "example.com")
+                flowdata.evidence.add("header.protocol", "ftp")
+                flowdata.process()
+        errors = [
+            entry["message"] for entry in logger.memory_log
+            if entry["level"] == "error"]
+
+        for variables in captured.variables:
+            self.assertNotIn("_url", variables)
+            self.assertFalse(variables["_updateEnabled"])
+        self.assertEqual(1, len(errors), errors)
+        self.assertIn("'ftp://example.com/json'", errors[0])
+
     def test_blank_host_gives_no_callback_and_is_not_logged(self):
         """A host made only of whitespace is not set, so the script has no
         callback and nothing is logged."""
@@ -232,5 +261,25 @@ def test_host_from_the_request_is_written_as_checked(self):
             "https://example.com/%27);alert(1);//json", rendered[0]["_url"])
 
 
+class JavaScriptBuilderProtocolSettingTests(unittest.TestCase):
+
+    @parameterized.expand([
+        ("ftp",), ("ws",), ("javascript",), ("https ",), ("http\u017f",),
+    ])
+    def test_invalid_configured_protocol_is_refused(self, protocol):
+        """A protocol in the settings is used for every request, so one the
+        script cannot send its request with is refused when the builder is
+        created."""
+
+        with self.assertRaises(ValueError):
+            javascriptbuilder.JavascriptBuilderElement({"protocol": protocol})
+
+    @parameterized.expand([("http",), ("https",), ("HTTPS",)])
+    def test_valid_configured_protocol_is_accepted(self, protocol):
+        element = javascriptbuilder.JavascriptBuilderElement(
+            {"protocol": protocol})
+        self.assertEqual(protocol, element.settings["_protocol"])
+
+
 if __name__ == "__main__":
     unittest.main()