#!/usr/bin/env python3 """Contract tests for the CVE-2026-109618 / -109619 edge mitigations. * templates/hap_listener.tpl denies multipart/form-data without a Content-Length (400), with AND without the Coraza SPOE block rendered. * coraza-spoa/overrides.conf carries the body rules 209610-209616. What these rules DO is measured elsewhere: the Coraza verdicts by WHP's scripts/tests/coraza-body-parse-harness (cve109618.go, run with HMB_OVERRIDES pointing at this checkout's overrides.conf), the HAProxy verdict by live curl against a host. This file pins the invariants that make those measurements stay true: * the deny is outside the `coraza_spoe_backend` conditional; * it keys on the Content-Length HEADER count, never req.body_size (see the H3 note in hap_listener.tpl); * every 2096xx id is inside 200000-209999 -- WHP's break-glass rule (991000001) removes exactly that range, so an id outside it would stay armed for a site in emergency mode; * no ctl:ruleEngine=On sits on a chain STARTER (Coraza runs it as soon as the first link matches -- the 920440 defect). Running ------- python3 scripts/test-multipart-cve-mitigation.py """ import logging import os import re import sys 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 CORAZA_OVERRIDES = os.path.join(MODULE_DIR, 'coraza-spoa', 'overrides.conf') ENGINE_LO, ENGINE_HI = 200000, 209999 # WHP_WAF_ENGINE_RANGE_LO..HI CVE_IDS = range(209610, 209617) def render(coraza): kw = dict(crt_path='/etc/haproxy/certs', suspension_enabled=True, coraza_spoe_backend='coraza-spoa:9000' if coraza else None, cluster_secret=None, syslog_target='172.18.0.1:514') return haproxy_manager.template_env.get_template('hap_listener.tpl').render(**kw) def active_lines(text): return [l.strip() for l in text.splitlines() if l.strip() and not l.strip().startswith('#')] DENY_RE = re.compile( r'^http-request deny deny_status 400 .*\bif is_multipart_form ' r'!\{ req\.hdr_cnt\(content-length\) gt 0 \}$') class ListenerDeny(unittest.TestCase): def check(self, coraza): lines = active_lines(render(coraza)) acl = [i for i, l in enumerate(lines) if l.startswith('acl is_multipart_form ')] deny = [i for i, l in enumerate(lines) if DENY_RE.match(l)] self.assertEqual(len(acl), 1, 'is_multipart_form acl missing or duplicated') self.assertEqual(len(deny), 1, 'multipart no-Content-Length deny missing or duplicated') self.assertEqual(lines[acl[0]], 'acl is_multipart_form req.fhdr(content-type) -m sub -i multipart/form-data') self.assertLess(acl[0], deny[0]) self.assertNotIn('body_size', lines[deny[0]]) # after the real-IP resolution, before any backend selection rule real_ip = max(i for i, l in enumerate(lines) if 'set-var(txn.real_ip)' in l) self.assertLess(real_ip, deny[0]) if coraza: spoe = [i for i, l in enumerate(lines) if l.startswith('http-request send-spoe-group')] self.assertTrue(spoe, 'coraza render has no send-spoe-group -- test is blind') self.assertLess(deny[0], spoe[0], 'deny must run before the WAF is consulted') def test_with_coraza(self): self.check(True) def test_without_coraza(self): self.check(False) @unittest.skipUnless(os.path.exists(CORAZA_OVERRIDES), 'coraza-spoa/overrides.conf is not shipped in this image') class OverrideRules(unittest.TestCase): def setUp(self): with open(CORAZA_OVERRIDES) as fh: text = fh.read() # Join backslash continuations so one directive is one string. joined = re.sub(r'\\\n\s*', '', text) self.rules = [l.strip() for l in joined.splitlines() if l.strip().startswith('SecRule') or l.strip().startswith('SecAction')] def ids(self): out = [] for r in self.rules: m = re.search(r'\bid:(\d+)', r) if m: out.append(int(m.group(1))) return out def test_every_cve_id_present_once(self): ids = self.ids() for rid in CVE_IDS: self.assertEqual(ids.count(rid), 1, f'{rid} missing or duplicated') def test_every_id_inside_the_break_glass_engine_range(self): # every rule that belongs to the mitigation, found by what it DOES # (its tag or its tx variables), not by its id cve = [int(re.search(r'\bid:(\d+)', r).group(1)) for r in self.rules if re.search(r'\bid:\d+', r) and ('whp/cve-2026-1096' in r or 'whp_cve109618' in r)] self.assertGreaterEqual(len(cve), 7, 'CVE rules not found -- the test is blind') self.assertTrue(cve, 'no CVE ids parsed -- the test is blind') for rid in cve: self.assertTrue(ENGINE_LO <= rid <= ENGINE_HI, rid) def test_deny_rules_flip_the_engine_for_themselves(self): denies = [r for r in self.rules if re.search(r'\bid:2096\d\d\b', r) and ',deny,' in r] self.assertEqual(sorted(int(re.search(r'id:(\d+)', r).group(1)) for r in denies), [209610, 209613, 209615, 209616]) for i, r in enumerate(self.rules): if r not in denies: continue if ',chain' in r: self.assertNotIn('ctl:ruleEngine', r, 'engine flip on a chain starter') self.assertIn('ctl:ruleEngine=On', self.rules[i + 1]) else: self.assertIn('ctl:ruleEngine=On', r) def test_no_chain_starter_anywhere_in_the_block_flips_the_engine(self): for r in self.rules: if re.search(r'\bid:2096\d\d\b', r) and ',chain' in r: self.assertNotIn('ctl:ruleEngine', r) if __name__ == '__main__': unittest.main(verbosity=2)