fix(haproxy): only honour proxy headers from trusted sources

Restrict CF-Connecting-IP, X-Real-IP and X-Forwarded-For to peers matching cloudflare_ips.list or trusted_proxies.list; other peers fall through to src. Adds a regression test pinning the strip-before-resolve ordering.
2026-08-13 13:28:31 -07:00
parent af9fb1d2f0
commit 79a1b84ca2
2 changed files with 110 additions and 0 deletions
@@ -0,0 +1,88 @@
+#!/usr/bin/env python3
+"""Regression tests for the trusted-proxy header gate in hap_listener.tpl.
+
+Why this file exists
+--------------------
+txn.real_ip was derived from CF-Connecting-IP / X-Real-IP / X-Forwarded-For
+with no check on the peer, so any direct client could assert any client IP.
+That variable drives rate limiting, the trusted-IP whitelist, the wp-login
+brute-force table and cookie challenge, the wp-json/batch/v1 virtual patch,
+IP blocking, and Coraza's src-ip -- so a spoofed header bypassed all of them.
+
+These tests pin the invariant that the header strip is rendered BEFORE any
+real-IP resolution. Ordering is the whole fix: a del-header emitted after the
+set-var chain would parse fine, validate fine, and do nothing.
+
+Running
+-------
+    python3 scripts/test-trusted-proxy-gate.py
+"""
+
+import os
+import re
+import sys
+import logging
+import tempfile
+import unittest
+
+MODULE_DIR = os.path.abspath(
+    os.environ.get('HAPROXY_MANAGER_DIR',
+                   os.path.join(os.path.dirname(os.path.abspath(__file__)), '..'))
+)
+os.chdir(MODULE_DIR)
+sys.path.insert(0, MODULE_DIR)
+
+_LOG_DIR = tempfile.mkdtemp(prefix='haproxy-mgr-test-logs-')
+_real_file_handler = logging.FileHandler
+logging.FileHandler = (
+    lambda filename, *a, **kw: _real_file_handler(
+        os.path.join(_LOG_DIR, os.path.basename(filename)), *a, **kw)
+)
+
+import haproxy_manager  # noqa: E402
+
+GATED_HEADERS = ('CF-Connecting-IP', 'X-Real-IP', 'X-Forwarded-For')
+
+
+def render_listener():
+    return haproxy_manager.template_env.get_template('hap_listener.tpl').render(
+        crt_path='/etc/haproxy/certs',
+        suspension_enabled=False,
+        coraza_spoe_backend=None,
+    )
+
+
+class TrustedProxyGate(unittest.TestCase):
+
+    def setUp(self):
+        self.cfg = render_listener()
+
+    def test_trusted_proxy_acl_is_defined(self):
+        self.assertIn('acl from_trusted_proxy src', self.cfg)
+        self.assertIn('/etc/haproxy/cloudflare_ips.list', self.cfg)
+        self.assertIn('/etc/haproxy/trusted_proxies.list', self.cfg)
+
+    def test_each_header_is_deleted_for_untrusted_peers(self):
+        for header in GATED_HEADERS:
+            with self.subTest(header=header):
+                pattern = (r'http-request\s+del-header\s+%s\s+if\s+!from_trusted_proxy'
+                           % re.escape(header))
+                self.assertRegex(self.cfg, pattern)
+
+    def test_strip_precedes_real_ip_resolution(self):
+        """The ordering invariant. A strip after the set-var chain is a no-op."""
+        last_strip = max(
+            self.cfg.index('del-header %s' % h) for h in GATED_HEADERS
+        )
+        first_setvar = self.cfg.index('set-var(txn.real_ip)')
+        self.assertLess(
+            last_strip, first_setvar,
+            'del-header rules must be rendered before set-var(txn.real_ip)')
+
+    def test_src_fallback_still_present(self):
+        """Direct clients must fall through to the real TCP peer."""
+        self.assertIn('set-var(txn.real_ip) src', self.cfg)
+
+
+if __name__ == '__main__':
+    unittest.main(verbosity=2)