CVE-2021-29431: Server-Side Request Forgery (SSRF) (CWE-918)
value: Sydent is a reference Matrix identity server. Sydent can be induced to send HTTP GET requests to internal systems, due to lack of parameter validation or IP address blacklisting. It is not possible to exfiltrate data or control request headers, but it might be possible to use the attack to perform an internal port enumeration. This issue has been addressed in in 9e57334, 8936925, 3d531ed, 0f00412. A potential workaround would be to use a firewall to ensure that Sydent cannot reach internal HTTP resources.
CWE description: The web server receives a URL or similar request from an upstream component and retrieves the contents of this URL, but it does not sufficiently ensure that the request is being sent to the expected destination.
CVSS v3 base score: 6.5
Published: 2021-04-15T21:15Z
Repository: https://github.com/matrix-org/sydent
Language: Python
Files changed: sydent/http/servlets/registerservlet.py, sydent/util/stringutils.py, tests/test_auth.py, tests/test_register.py, tests/test_util.py
Commit date: 2021-03-24 13:50:40 +0000
Commit message: Rework hostname validation to make port checking stricter.

Instead of using a regex to validate the entire hostname + port
combination, we now split the hostname into components and check each
component separately. This makes the regex a bit simpler and allows us
to validate the port number better, including that it belongs to the
valid range.

Diff: diff --git a/sydent/http/servlets/registerservlet.py b/sydent/http/servlets/registerservlet.py
index ebd07cc..21d79af 100644
--- a/sydent/http/servlets/registerservlet.py
+++ b/sydent/http/servlets/registerservlet.py
@@ -23,11 +23,11 @@ import json
 from six.moves import urllib
 
 from sydent.http.servlets import get_args, jsonwrap, deferjsonwrap, send_cors
 from sydent.http.httpclient import FederationHttpClient
 from sydent.users.tokens import issueToken
-
+from sydent.util.stringutils import is_valid_hostname
 
 logger = logging.getLogger(__name__)
 
 
 class RegisterServlet(Resource):
@@ -45,13 +45,24 @@ class RegisterServlet(Resource):
         """
         send_cors(request)
 
         args = get_args(request, ('matrix_server_name', 'access_token'))
 
+        hostname = args['matrix_server_name'].lower()
+
+        if not is_valid_hostname(hostname):
+            request.setResponseCode(400)
+            return {
+                'errcode': 'M_INVALID_PARAM',
+                'error': 'matrix_server_name must be a valid hostname'
+            }
+
         result = yield self.client.get_json(
-            "matrix://%s/_matrix/federation/v1/openid/userinfo?access_token=%s" % (
-                args['matrix_server_name'], urllib.parse.quote(args['access_token']),
+            "matrix://%s/_matrix/federation/v1/openid/userinfo?access_token=%s"
+            % (
+                hostname,
+                urllib.parse.quote(args['access_token']),
             ),
             1024 * 5,
         )
         if 'sub' not in result:
             raise Exception("Invalid response from homeserver")
diff --git a/sydent/util/stringutils.py b/sydent/util/stringutils.py
index e41ff66..d2f3732 100644
--- a/sydent/util/stringutils.py
+++ b/sydent/util/stringutils.py
@@ -15,16 +15,56 @@
 import re
 
 # https://matrix.org/docs/spec/client_server/r0.6.0#post-matrix-client-r0-register-email-requesttoken
 client_secret_regex = re.compile(r"^[0-9a-zA-Z\.\=\_\-]+$")
 
+# hostname/domain name + optional port
+# https://regex101.com/r/OyN1lg/2
+hostname_regex = re.compile(
+    r"^(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?)(?:\.[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?)*$",
+    flags=re.IGNORECASE)
+
 
 def is_valid_client_secret(client_secret):
     """Validate that a given string matches the client_secret regex defined by the spec
 
     :param client_secret: The client_secret to validate
-    :type client_secret: unicode
+    :type client_secret: str
 
     :return: Whether the client_secret is valid
     :rtype: bool
     """
     return client_secret_regex.match(client_secret) is not None
+
+
+def is_valid_hostname(string: str) -> bool:
+    """Validate that a given string is a valid hostname or domain name, with an
+    optional port number.
+
+    For domain names, this only validates that the form is right (for
+    instance, it doesn't check that the TLD is valid). If a port is
+    specified, it has to be a valid port number.
+
+    :param string: The string to validate
+    :type string: str
+
+    :return: Whether the input is a valid hostname
+    :rtype: bool
+    """
+
+    host_parts = string.split(":", 1)
+
+    if len(host_parts) == 1:
+        return hostname_regex.match(string) is not None
+    else:
+        host, port = host_parts
+        valid_hostname = hostname_regex.match(host) is not None
+
+        try:
+            port_num = int(port)
+            valid_port = (
+                port == str(port_num)  # exclude things like '08090' or ' 8090'
+                and 1 <= port_num < 65536
+        except ValueError:
+            valid_port = False
+
+        return valid_hostname and valid_port
diff --git a/tests/test_auth.py b/tests/test_auth.py
index 7d25945..71f5a51 100644
--- a/tests/test_auth.py
+++ b/tests/test_auth.py
@@ -42,11 +42,11 @@ class AuthTestCase(unittest.TestCase):
         )
 
         self.sydent.db.commit()
 
     def test_can_read_token_from_headers(self):
-        """Tests that Sydent correct extracts an auth token from request headers"""
+        """Tests that Sydent correctly extracts an auth token from request headers"""
         self.sydent.run()
 
         request, _ = make_request(
             self.sydent.reactor, "GET", "/_matrix/identity/v2/hash_details"
         )
@@ -57,11 +57,11 @@ class AuthTestCase(unittest.TestCase):
         token = tokenFromRequest(request)
 
         self.assertEqual(token, self.test_token)
 
     def test_can_read_token_from_query_parameters(self):
-        """Tests that Sydent correct extracts an auth token from query parameters"""
+        """Tests that Sydent correctly extracts an auth token from query parameters"""
         self.sydent.run()
 
         request, _ = make_request(
             self.sydent.reactor, "GET",
             "/_matrix/identity/v2/hash_details?access_token=" + self.test_token
diff --git a/tests/test_register.py b/tests/test_register.py
new file mode 100644
index 0000000..abc8c16
--- /dev/null
+++ b/tests/test_register.py
@@ -0,0 +1,45 @@
+# -*- coding: utf-8 -*-
+
+# Copyright 2021 The Matrix.org Foundation C.I.C.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from twisted.trial import unittest
+
+from tests.utils import make_request, make_sydent
+
+
+class RegisterTestCase(unittest.TestCase):
+    """Tests Sydent's register servlet"""
+    def setUp(self):
+        # Create a new sydent
+        self.sydent = make_sydent()
+
+    def test_sydent_rejects_invalid_hostname(self):
+        """Tests that the /register endpoint rejects an invalid hostname passed as matrix_server_name"""
+        self.sydent.run()
+
+        bad_hostname = "example.com#"
+
+        request, channel = make_request(
+            self.sydent.reactor,
+            "POST",
+            "/_matrix/identity/v2/account/register",
+            content={
+                "matrix_server_name": bad_hostname,
+                "access_token": "foo"
+            })
+
+        request.render(self.sydent.servlets.registerServlet)
+
+        self.assertEqual(channel.code, 400)
diff --git a/tests/test_util.py b/tests/test_util.py
new file mode 100644
index 0000000..7c9a011
--- /dev/null
+++ b/tests/test_util.py
@@ -0,0 +1,26 @@
+from twisted.trial import unittest
+from sydent.util.stringutils import is_valid_hostname
+
+
+class UtilTests(unittest.TestCase):
+    """Tests Sydent utility functions."""
+    def test_is_valid_hostname(self):
+        """Tests that the is_valid_hostname function accepts only valid
+        hostnames (or domain names), with optional port number.
+        """
+
+        self.assertTrue(is_valid_hostname("example.com"))
+        self.assertTrue(is_valid_hostname("EXAMPLE.COM"))
+        self.assertTrue(is_valid_hostname("ExAmPlE.CoM"))
+        self.assertTrue(is_valid_hostname("example.com:4242"))
+        self.assertTrue(is_valid_hostname("localhost"))
+        self.assertTrue(is_valid_hostname("localhost:9000"))
+        self.assertTrue(is_valid_hostname("a.b:1234"))
+
+        self.assertFalse(is_valid_hostname("example.com:65536"))
+        self.assertFalse(is_valid_hostname("example.com:0"))
+        self.assertFalse(is_valid_hostname("example.com:a"))
+        self.assertFalse(is_valid_hostname("example.com:04242"))
+        self.assertFalse(is_valid_hostname("example.com: 4242"))
+        self.assertFalse(is_valid_hostname("example.com/example.com"))
+        self.assertFalse(is_valid_hostname("example.com#example.com"))

Vulnerable code: 
            "matrix://%s/_matrix/federation/v1/openid/userinfo?access_token=%s" % (
                args['matrix_server_name'], urllib.parse.quote(args['access_token']),
    :type client_secret: unicode
        """Tests that Sydent correct extracts an auth token from request headers"""
        """Tests that Sydent correct extracts an auth token from query parameters"""
Fixed code: from sydent.util.stringutils import is_valid_hostname
        hostname = args['matrix_server_name'].lower()

        if not is_valid_hostname(hostname):
            request.setResponseCode(400)
            return {
                'errcode': 'M_INVALID_PARAM',
                'error': 'matrix_server_name must be a valid hostname'
            }

            "matrix://%s/_matrix/federation/v1/openid/userinfo?access_token=%s"
            % (
                hostname,
                urllib.parse.quote(args['access_token']),
# hostname/domain name + optional port
# https://regex101.com/r/OyN1lg/2
hostname_regex = re.compile(
    r"^(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?)(?:\.[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?)*$",
    flags=re.IGNORECASE)

    :type client_secret: str


def is_valid_hostname(string: str) -> bool:
    """Validate that a given string is a valid hostname or domain name, with an
    optional port number.

    For domain names, this only validates that the form is right (for
    instance, it doesn't check that the TLD is valid). If a port is
    specified, it has to be a valid port number.

    :param string: The string to validate
    :type string: str

    :return: Whether the input is a valid hostname
    :rtype: bool
    """

    host_parts = string.split(":", 1)

    if len(host_parts) == 1:
        return hostname_regex.match(string) is not None
    else:
        host, port = host_parts
        valid_hostname = hostname_regex.match(host) is not None

        try:
            port_num = int(port)
            valid_port = (
                port == str(port_num)  # exclude things like '08090' or ' 8090'
                and 1 <= port_num < 65536
        except ValueError:
            valid_port = False

        return valid_hostname and valid_port
        """Tests that Sydent correctly extracts an auth token from request headers"""
        """Tests that Sydent correctly extracts an auth token from query parameters"""
# -*- coding: utf-8 -*-

# Copyright 2021 The Matrix.org Foundation C.I.C.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from twisted.trial import unittest

from tests.utils import make_request, make_sydent


class RegisterTestCase(unittest.TestCase):
    """Tests Sydent's register servlet"""
    def setUp(self):
        # Create a new sydent
        self.sydent = make_sydent()

    def test_sydent_rejects_invalid_hostname(self):
        """Tests that the /register endpoint rejects an invalid hostname passed as matrix_server_name"""
        self.sydent.run()

        bad_hostname = "example.com#"

        request, channel = make_request(
            self.sydent.reactor,
            "POST",
            "/_matrix/identity/v2/account/register",
            content={
                "matrix_server_name": bad_hostname,
                "access_token": "foo"
            })

        request.render(self.sydent.servlets.registerServlet)

        self.assertEqual(channel.code, 400)
from twisted.trial import unittest
from sydent.util.stringutils import is_valid_hostname


class UtilTests(unittest.TestCase):
    """Tests Sydent utility functions."""
    def test_is_valid_hostname(self):
        """Tests that the is_valid_hostname function accepts only valid
        hostnames (or domain names), with optional port number.
        """

        self.assertTrue(is_valid_hostname("example.com"))
        self.assertTrue(is_valid_hostname("EXAMPLE.COM"))
        self.assertTrue(is_valid_hostname("ExAmPlE.CoM"))
        self.assertTrue(is_valid_hostname("example.com:4242"))
        self.assertTrue(is_valid_hostname("localhost"))
        self.assertTrue(is_valid_hostname("localhost:9000"))
        self.assertTrue(is_valid_hostname("a.b:1234"))

        self.assertFalse(is_valid_hostname("example.com:65536"))
        self.assertFalse(is_valid_hostname("example.com:0"))
        self.assertFalse(is_valid_hostname("example.com:a"))
        self.assertFalse(is_valid_hostname("example.com:04242"))
        self.assertFalse(is_valid_hostname("example.com: 4242"))
        self.assertFalse(is_valid_hostname("example.com/example.com"))
        self.assertFalse(is_valid_hostname("example.com#example.com"))