improve robustness of host comparision for wss:// Origin headers

see: https://github.com/ansible/tower/issues/2647

awx/main/conf.py

@@ -101,6 +101,17 @@ register(
     category_slug='system',
 )
 
+register(
+    'WEBSOCKET_ORIGIN_WHITELIST',
+    field_class=fields.StringListField,
+    label=_('Websocket Origin Whitelist'),
+    help_text=_("If Tower is behind a reverse proxy/load balancer, use this setting "
+                "to whitelist hostnames which represent trusted Origin hostnames from which "
+                "Tower should allow websocket connections."),
+    category=_('System'),
+    category_slug='system',
+)
+
 
 def _load_default_license_from_file():
     try:

awx/main/consumers.py

@@ -4,7 +4,9 @@ import logging
 from channels import Group
 from channels.auth import channel_session_user_from_http, channel_session_user
 from channels.exceptions import DenyConnection
+from six.moves.urllib.parse import urlparse
 
+from django.utils.http import is_same_domain
 from django.conf import settings
 from django.core.serializers.json import DjangoJSONEncoder
 
@@ -18,12 +20,42 @@ def discard_groups(message):
             Group(group).discard(message.reply_channel)
 
 
+def origin_is_valid(message, trusted_values):
+    origin = dict(message.content.get('headers', {})).get('origin', '')
+    for trusted in trusted_values:
+        try:
+            client = urlparse(origin)
+            trusted = urlparse(trusted)
+        except (AttributeError, ValueError):
+            # if we can't parse the origin header, fall back to the else block
+            pass
+        else:
+            # if we _can_ parse the origin header, verify that it's trusted
+            if (
+                trusted.scheme == client.scheme and
+                is_same_domain(client.netloc, trusted.netloc)
+            ):
+                # the provided Origin matches at least _one_ whitelisted host,
+                # break out and accept the connection
+                break
+    else:
+        logger.error((
+            "ws:// origin header mismatch {} not in {}; consider adding {} to "
+            "settings.WEBSOCKET_ORIGIN_WHITELIST if it's a trusted host."
+        ).format(origin, trusted_values, origin))
+        return False
+    return True
+
+
+
 @channel_session_user_from_http
 def ws_connect(message):
-    origin = dict(message.content.get('headers', {})).get('origin')
+    if not origin_is_valid(
-    if settings.DEBUG is False and origin != settings.TOWER_URL_BASE:
+        message,
-        logger.error("ws:// origin header mismatch {} != {}".format(origin, settings.TOWER_URL_BASE))
+        [settings.TOWER_URL_BASE] + settings.WEBSOCKET_ORIGIN_WHITELIST
+    ):
         raise DenyConnection()
+
     message.reply_channel.send({"accept": True})
     message.content['method'] = 'FAKE'
     if message.user.is_authenticated():

awx/main/tests/unit/test_consumers.py

@@ -0,0 +1,33 @@
+from collections import namedtuple
+
+import pytest
+
+from awx.main.consumers import origin_is_valid
+
+
+def _msg(origin):
+    return namedtuple('message', ('content',))({
+        'headers': [
+            ('origin', origin)
+        ]
+    })
+
+
+@pytest.mark.parametrize('origin, trusted, valid', [
+    ('https://tower.example.org', ['https://tower.example.org'], True),           # exact match
+    ('https://tower.example.org/', ['https://tower.example.org'], True),          # trailing slash match
+    ('https://tower.example.org', ['https://.example.org'], True),                # wildcard match
+    ('https://proxy.tower.example.org', ['https://.tower.example.org'], True),    # complex wildcard match
+    ('', ['https://tower.example.org'], False),                                   # origin header empty
+    (None, ['https://tower.example.org'], False),                                 # origin header unset
+    ('https://[\">[', ['https://tower.example.org'], False),                      # origin header garbage
+    ('file:///bad.html', ['https://tower.example.org'], False),                   # file:// origin blocked
+    ('http://tower.example.org', ['https://tower.example.org'], False),           # http != https
+    ('https://tower.example.org:443', ['https://tower.example.org:8043'], False), # port mismatch
+    ('https://evil.example.com', ['https://tower.example.org'], False),           # domain mismatch
+    ('https://tower.example.org', [], False),                                     # no trusted hosts
+    ('https://a', ['https://a', 'https://b'], True),                              # multiple with a match
+    ('https://evil', ['https://a', 'https://b'], False),                          # multiple no match
+])
+def test_trusted_origin(origin, trusted, valid):
+    assert origin_is_valid(_msg(origin), trusted) is valid

awx/settings/defaults.py

@@ -166,6 +166,11 @@ REMOTE_HOST_HEADERS = ['REMOTE_ADDR', 'REMOTE_HOST']
 # REMOTE_HOST_HEADERS will be trusted unconditionally')
 PROXY_IP_WHITELIST = []
 
+# If Tower is behind a reverse proxy/load balancer, use this setting
+# to whitelist hostnames which represent trusted Origin hostnames from which
+# Tower should allow websocket connections.
+WEBSOCKET_ORIGIN_WHITELIST = []
+
 # Note: This setting may be overridden by database settings.
 STDOUT_MAX_BYTES_DISPLAY = 1048576