haproxy-manager-base/scripts/test-multipart-cve-mitigation.py

#!/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)