Only trust last entry in x-forwarded-for by default (ansible#575)

Author: john-westcott-iv

ansible_base/lib/utils/requests.py

@@ -15,6 +15,15 @@ def get_remote_host(request: HttpRequest) -> Optional[str]:
     return value[0] if value else None
 
 
+def split_header(value: str) -> list[str]:
+    values = []
+    for a_value in value.split(','):
+        a_value = a_value.strip()
+        if a_value:
+            values.append(a_value)
+    return values
+
+
 def get_remote_hosts(request: HttpRequest, get_first_only: bool = False) -> list[str]:
     '''
     Get all IPs from the allowed headers
@@ -27,24 +36,32 @@ def get_remote_hosts(request: HttpRequest, get_first_only: bool = False) -> list
 
     headers = get_setting('REMOTE_HOST_HEADERS', ['REMOTE_ADDR', 'REMOTE_HOST'])
 
+    for header in headers:
+        for value in split_header(request.META.get(header, '')):
+            remote_hosts.append(value)
+
     # If we are connected to from a trusted proxy then we can add some additional headers
     try:
         if 'HTTP_X_TRUSTED_PROXY' in request.META:
             if validate_x_trusted_proxy_header(request.META['HTTP_X_TRUSTED_PROXY']):
-                headers.insert(0, 'HTTP_X_FORWARDED_FOR')
-                headers.insert(0, 'HTTP_X_ENVOY_EXTERNAL_ADDRESS')
+                # The last entry in x-forwarded-for from envoy can be trusted implicitly
+                # https://www.envoyproxy.io/docs/envoy/latest/configuration/http/http_conn_man/headers#x-forwarded-for
+                values = split_header(request.META.get('HTTP_X_FORWARDED_FOR', ''))
+                if values:
+                    remote_hosts.insert(0, values[-1])
+
+                # x-envoy-external-address can always be trusted when coming from envoy
+                # https://www.envoyproxy.io/docs/envoy/latest/configuration/http/http_conn_man/headers#x-envoy-external-address
+                for value in reversed(split_header(request.META.get('HTTP_X_ENVOY_EXTERNAL_ADDRESS', ''))):
+                    remote_hosts.insert(0, value)
             else:
                 logger.error("Unable to use headers from trusted proxy because shared secret was invalid!")
     except Exception:
         logger.exception("Failed to validate HTTP_X_TRUSTED_PROXY")
 
-    for header in headers:
-        for value in request.META.get(header, '').split(','):
-            value = value.strip()
-            if value:
-                if get_first_only:
-                    return [value]
-                remote_hosts.append(value)
+    if get_first_only and len(remote_hosts) > 0:
+        return [remote_hosts[0]]
+
     return remote_hosts
 
 

test_app/tests/lib/utils/test_requests.py

@@ -52,7 +52,7 @@ def test_get_remote_hosts_no_headers():
             },
             {'REMOTE_ADDR': '9.10.11.12', 'REMOTE_HOST': 'localhost, example.com'},
             False,
-            ['5.6.7.8', '9.10.11.12', '1.2.3.4', '5.6.7.8', '9.10.11.12', 'localhost', 'example.com'],
+            ['5.6.7.8', '9.10.11.12', '5.6.7.8', '9.10.11.12', 'localhost', 'example.com'],
         ),
         # Complicated example w/o the trusted_proxy header
         (
@@ -100,7 +100,7 @@ def test_get_remote_hosts_alternate_headers_behind_trusted_proxy(rsa_keypair):
         request = RequestFactory().get('/hello', headers=headers)
         with override_settings(REMOTE_HOST_HEADERS=['HTTP_X_JOHNS_HEADER']):
             remote_hosts = get_remote_hosts(request, get_first_only=False)
-            assert remote_hosts == ['5.6.7.8', '9.10.11.12', '1.2.3.4', '5.6.7.8', 'a.b.c.d']
+            assert remote_hosts == ['5.6.7.8', '9.10.11.12', '5.6.7.8', 'a.b.c.d']
 
 
 @mock.patch("ansible_base.lib.utils.requests.validate_x_trusted_proxy_header", side_effect=Exception)