toolbox.py

# -*- coding: utf-8 -*-
"""
POSC工具箱
打包：pyinstaller -F -w toolbox.py
"""
import hashlib
import json
import mmap
import re
import sys
import time
from datetime import datetime
from pathlib import Path
from threading import Thread

import requests
from PyQt5.QtCore import QThread, pyqtSignal
from PyQt5.QtWidgets import (QApplication, QMainWindow, QHBoxLayout,
                             QVBoxLayout, QWidget, QPushButton,
                             QTextEdit, QLabel, QMessageBox,
                             QDialog, QFormLayout, QDialogButtonBox,
                             QLineEdit, QComboBox)

# ===== 通用 =====
def exe_dir():
    return Path(sys.executable).parent if getattr(sys, 'frozen', False) else Path(__file__).parent

LOG_FILE = exe_dir() / "抓包.txt"
POST_HISTORY_FILE = exe_dir() / "post测试.txt"

def log_write(msg: str):
    with open(LOG_FILE, "a", encoding="utf-8") as f:
        f.write(f"{datetime.now()}  {msg}\n")

def load_post_history():
    if POST_HISTORY_FILE.exists():
        with open(POST_HISTORY_FILE, "r", encoding="utf-8") as f:
            lines = f.read().splitlines()
            url = lines[0] if lines else "http://iotapi.abupdate.com/product/"
            data = "\n".join(lines[1:]) if len(lines) > 1 else json.dumps({
                "timestamp": "", "sign": "", "mid": "",
                "productId": "", "version": "", "networkType": "WIFI"
            }, ensure_ascii=False)
            return url, data
    return ("http://iotapi.abupdate.com/product/", json.dumps({
        "timestamp": "", "sign": "", "mid": "",
        "productId": "", "version": "", "networkType": "WIFI"
    }, ensure_ascii=False))

def save_post_history(url: str, data: str, resp: str):
    with open(POST_HISTORY_FILE, "w", encoding="utf-8") as f:
        f.write(f"{url}\n{data}\n\n---POST 返回---\n{resp}\n")

# ===== 管理员 =====
def is_admin():
    try:
        import ctypes
        return ctypes.windll.shell32.IsUserAnAdmin() != 0
    except Exception:
        return False

def elevate():
    try:
        import ctypes
        ctypes.windll.shell32.ShellExecuteW(
            None, "runas", sys.executable, " ".join(sys.argv), None, 1)
    except Exception:
        QMessageBox.critical(None, "提权失败", "无法获得管理员权限，抓包功能将不可用。")

# ===== 抓包线程 =====
class HttpSniffWorker(QThread):
    log = pyqtSignal(str)
    finished = pyqtSignal()

    def __init__(self, iface: str):
        super().__init__()
        self.iface = iface
        self._running = True

    def stop(self):
        self._running = False

    def run(self):
        import scapy.all as scapy
        (exe_dir() / "抓包.txt").write_text("", encoding="utf-8")
        log_write("=== 抓包会话开始 ===")

        def handle(pkt):
            if not self._running:
                return
            if pkt.haslayer(scapy.TCP) and pkt.haslayer(scapy.Raw):
                payload = bytes(pkt[scapy.Raw]).decode(errors="ignore")
                if ("POST" in payload or "GET" in payload) and "HTTP/" in payload:
                    self.log.emit(payload.strip())
                    log_write(payload.strip())

        try:
            scapy.sniff(iface=self.iface,
                        filter="tcp",
                        prn=handle,
                        stop_filter=lambda _: not self._running)
        except Exception as e:
            err = f"抓包错误：{e}"
            self.log.emit(err)
            log_write(err)
        finally:
            log_write("=== 抓包会话结束 ===")
            self.finished.emit()

# ===== POST 测试 =====
class PostTestDialog(QDialog):
    def __init__(self, parent=None):
        super().__init__(parent)
        self.setWindowTitle("POST 测试")
        self.resize(450, 320)
        url, data = load_post_history()
        try:
            history = json.loads(data)
        except Exception:
            history = {}
        self.url_edit  = QLineEdit(history.get("url", "http://iotapi.abupdate.com/product/"))
        self.ts_edit   = QLineEdit(str(history.get("timestamp", "")))
        self.sign_edit = QLineEdit(str(history.get("sign", "")))
        self.mid_edit  = QLineEdit(str(history.get("mid", "")))
        self.pid_edit  = QLineEdit(str(history.get("productId", "")))
        self.ver_edit  = QLineEdit(str(history.get("version", "")))
        self.net_label = QLabel("WIFI")
        self.net_label.setEnabled(False)

        form = QFormLayout(self)
        form.addRow("URL:",       self.url_edit)
        form.addRow("timestamp:", self.ts_edit)
        form.addRow("sign:",      self.sign_edit)
        form.addRow("mid:",       self.mid_edit)
        form.addRow("productId:", self.pid_edit)
        form.addRow("version:",   self.ver_edit)
        form.addRow("networkType:", self.net_label)

        btns = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel)
        form.addRow(btns)
        btns.accepted.connect(self.accept)
        btns.rejected.connect(self.reject)

    def values(self):
        data = {
            "timestamp":   int(self.ts_edit.text().strip()),
            "sign":        self.sign_edit.text().strip(),
            "mid":         self.mid_edit.text().strip(),
            "productId":   self.pid_edit.text().strip(),
            "version":     self.ver_edit.text().strip(),
            "networkType": "WIFI"
        }
        return self.url_edit.text().strip(), data

    def accept(self):
        if not self.url_edit.text().strip():
            QMessageBox.warning(self, "提示", "URL 不能为空"); return
        ts = self.ts_edit.text().strip()
        if not ts:
            QMessageBox.warning(self, "提示", "timestamp 不能为空"); return
        try:
            int(ts)
        except ValueError:
            QMessageBox.warning(self, "提示", "timestamp 必须是整数"); return
        if not self.sign_edit.text().strip():
            QMessageBox.warning(self, "提示", "sign 不能为空"); return
        if not self.mid_edit.text().strip():
            QMessageBox.warning(self, "提示", "mid 不能为空"); return
        if not self.pid_edit.text().strip():
            QMessageBox.warning(self, "提示", "productId 不能为空"); return
        if not self.ver_edit.text().strip():
            QMessageBox.warning(self, "提示", "version 不能为空"); return
        super().accept()

# ===== 计算密码 =====
class CalcSignDialog(QDialog):
    def __init__(self, parent=None):
        super().__init__(parent)
        self.setWindowTitle("计算密码")
        self.resize(400, 200)
        self.src_edit = QLineEdit()
        self.src_edit.setPlaceholderText("请输入原密码")
        self.method_combo = QComboBox()
        self.method_combo.addItems(["MD5(+\\n)", "SHA256"])
        self.result_edit = QLineEdit()
        self.result_edit.setReadOnly(True)

        form = QFormLayout(self)
        form.addRow("原密码:",  self.src_edit)
        form.addRow("算法:",   self.method_combo)
        form.addRow("结果:",   self.result_edit)

        btn_calc = QPushButton("计算")
        btn_calc.clicked.connect(self.calc_hash)
        form.addRow(btn_calc)

    def calc_hash(self):
        src = self.src_edit.text()
        if not src:
            QMessageBox.warning(self, "提示", "请输入原密码")
            return
        mode = self.method_combo.currentText()
        if mode == "MD5(+\\n)":
            digest = hashlib.md5((src + "\n").encode("utf-8")).hexdigest()
        else:
            digest = hashlib.sha256(src.encode("utf-8")).hexdigest()
        self.result_edit.setText(digest)

# ===== 读码 =====
class ReadCodeDialog(QDialog):
    def __init__(self, parent=None):
        super().__init__(parent)
        self.setWindowTitle("读码")
        self.resize(600, 240)
        self.img_combo = QComboBox()
        self.populate_img()

        self.result_edit = QTextEdit()
        self.result_edit.setReadOnly(True)

        self.btn_read = QPushButton("确认")
        self.btn_read.clicked.connect(self.read_code)

        layout = QVBoxLayout(self)
        layout.addWidget(QLabel("IMG 文件:"))
        layout.addWidget(self.img_combo)
        layout.addWidget(QLabel("扫描结果:"))
        layout.addWidget(self.result_edit)
        layout.addWidget(self.btn_read)

    def populate_img(self):
        img_dir = exe_dir() / "img"
        img_dir.mkdir(exist_ok=True)
        for f in img_dir.glob("*.img"):
            self.img_combo.addItem(str(f))

    def read_code(self):
        img_path = self.img_combo.currentText()
        if not img_path:
            return
        try:
            with open(img_path, "rb") as f:
                data = f.read().decode(errors="ignore")

            hashes = []
            # SHA256 脚本
            for m in re.finditer(r'#\s*([a-f0-9]{64})\s*-', data, re.MULTILINE):
                hashes.append(("SHA256", m.group(1)))
            # MD5 脚本
            for m in re.finditer(r'=\s*"([a-f0-9]{32})\s*-"', data, re.MULTILINE):
                hashes.append(("MD5", m.group(1)))

            if not hashes:
                self.result_edit.setPlainText("未检测到密码")
            else:
                self.result_edit.setPlainText("\n".join([f"{t}: {h}" for t, h in hashes]))
        except Exception as e:
            self.result_edit.setPlainText(f"读取失败：{e}")


# ===== 改包（先读码→再替换） =====
class BinPatchDialog(QDialog):
    def __init__(self, parent=None):
        super().__init__(parent)
        self.setWindowTitle("改包")
        self.resize(600, 260)

        self.img_combo = QComboBox()
        self.populate_img()

        self.old_combo = QComboBox()
        self.old_combo.setEditable(True)
        self.old_combo.setPlaceholderText("请选择或输入旧值")

        self.new_edit = QLineEdit()
        self.new_edit.setPlaceholderText("新 MD5 / SHA256")

        self.btn_scan = QPushButton("扫描旧值")
        self.btn_scan.clicked.connect(self.scan_hashes)

        form = QFormLayout(self)
        form.addRow("IMG 文件:", self.img_combo)
        form.addRow(self.btn_scan)
        form.addRow("旧值:", self.old_combo)
        form.addRow("新值:", self.new_edit)

        btns = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel)
        form.addRow(btns)
        btns.accepted.connect(self.run_patch)
        btns.rejected.connect(self.reject)

    # populate_img 与 scan_hashes 保持原样，无需改动
    def populate_img(self):
        img_dir = exe_dir() / "img"
        img_dir.mkdir(exist_ok=True)
        for f in img_dir.glob("*.img"):
            self.img_combo.addItem(str(f))

    def scan_hashes(self):
        img_path = Path(self.img_combo.currentText())
        if not img_path.exists():
            return
        try:
            data = img_path.read_text(encoding="utf-8", errors="ignore")
            hashes = []
            # SHA256
            for m in re.finditer(r'#\s*([a-f0-9]{64})\s*-', data, re.MULTILINE):
                hashes.append(m.group(1))
            # MD5
            for m in re.finditer(r'=\s*"([a-f0-9]{32})\s*-"', data, re.MULTILINE):
                hashes.append(m.group(1))

            self.old_combo.clear()
            self.old_combo.addItems(hashes)
            if not hashes:
                QMessageBox.information(self, "提示", "未找到旧值")
        except Exception as e:
            QMessageBox.critical(self, "失败", str(e))

    # 关键修复：改为文本级替换
    def run_patch(self):
        old_hex = self.old_combo.currentText().strip().lower()
        new_hex = self.new_edit.text().strip().lower()
        img_path = Path(self.img_combo.currentText())

        if not old_hex or not new_hex:
            QMessageBox.warning(self, "提示", "旧值和新值不能为空")
            return
        if len(old_hex) != len(new_hex) or len(old_hex) not in (32, 64):
            QMessageBox.warning(self, "提示", "长度必须是 32 (MD5) 或 64 (SHA256)")
            return

        try:
            text = img_path.read_text(encoding="utf-8", errors="ignore")

            # 构造正则，匹配完整旧文本（含前后固定格式）
            pattern = None
            if len(old_hex) == 64:       # SHA256
                pattern = rf'(#\s*)({re.escape(old_hex)})(\s*-)'
            elif len(old_hex) == 32:     # MD5
                pattern = rf'(=\s*")({re.escape(old_hex)})(\s*-")'

            match = re.search(pattern, text)
            if not match:
                raise RuntimeError("未找到目标值")

            # 保持原格式，仅替换哈希部分
            new_text = match.group(1) + new_hex + match.group(3)
            old_full = match.group(0)
            new_data = text.replace(old_full, new_text)

            img_path.write_text(new_data, encoding="utf-8")
            QMessageBox.information(self, "成功", "替换完成！")
        except Exception as e:
            QMessageBox.critical(self, "失败", str(e))

# ===== 主窗口 =====
class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("POSC工具箱")
        self.resize(900, 600)

        # 左侧日志
        self.log_area = QTextEdit()
        self.log_area.setReadOnly(True)

        # 右侧控制面板
        self.iface_combo = QComboBox()
        self.btn_refresh = QPushButton("刷新网卡")
        self.btn_refresh.clicked.connect(self.refresh_interfaces)

        self.btn_capture = QPushButton("抓包")
        self.btn_capture.setCheckable(True)
        self.btn_capture.clicked.connect(self.toggle_capture)

        self.btn_post = QPushButton("POST 测试")
        self.btn_post.clicked.connect(self.open_post)

        self.btn_calc = QPushButton("计算密码")
        self.btn_calc.clicked.connect(self.open_calc)

        self.btn_read = QPushButton("读码")
        self.btn_read.clicked.connect(self.open_read)

        self.btn_patch = QPushButton("改包")
        self.btn_patch.clicked.connect(self.open_patch)

        ctrl = QVBoxLayout()
        ctrl.addWidget(QLabel("网卡："))
        ctrl.addWidget(self.iface_combo)
        ctrl.addWidget(self.btn_refresh)
        ctrl.addWidget(self.btn_capture)
        ctrl.addWidget(self.btn_post)
        ctrl.addWidget(self.btn_calc)
        ctrl.addWidget(self.btn_read)
        ctrl.addWidget(self.btn_patch)
        ctrl.addStretch()

        main_layout = QHBoxLayout()
        main_layout.addWidget(self.log_area, stretch=2)
        main_layout.addLayout(ctrl, stretch=1)

        central = QWidget()
        central.setLayout(main_layout)
        self.setCentralWidget(central)
        self.refresh_interfaces()

    def refresh_interfaces(self):
        try:
            import scapy.all as scapy
            names = [scapy.conf.ifaces.get(i, {}).name or str(i) for i in scapy.get_if_list()]
            self.iface_combo.clear()
            self.iface_combo.addItems(names)
        except Exception as e:
            self.log_area.append(f"刷新网卡失败：{e}")

    def toggle_capture(self):
        if self.btn_capture.isChecked():
            iface = self.iface_combo.currentText()
            if not iface:
                QMessageBox.warning(self, "提示", "请选择网卡！")
                self.btn_capture.setChecked(False)
                return
            try:
                import scapy.all as scapy
                scapy.sniff(iface=iface, count=0, timeout=0.5)
            except PermissionError:
                QMessageBox.information(self, "需要管理员", "正在申请管理员权限…")
                elevate()
                QApplication.quit()
                return
            except Exception as e:
                QMessageBox.critical(self, "错误", str(e))
                self.btn_capture.setChecked(False)
                return
            self.worker = HttpSniffWorker(iface)
            self.worker.log.connect(self.log_area.append)
            self.worker.finished.connect(lambda: self.btn_capture.setChecked(False))
            self.worker.start()
            self.btn_capture.setText("停止抓包")
            self.log_area.append(f"[{datetime.now()}] 开始抓包：{iface}")
        else:
            if self.worker and self.worker.isRunning():
                self.worker.stop()
                self.worker.wait()
            self.btn_capture.setText("抓包")
            self.log_area.append(f"[{datetime.now()}] 抓包已停止")

    def open_post(self):
        dlg = PostTestDialog(self)
        if dlg.exec_() != QDialog.Accepted:
            return
        url, data_dict = dlg.values()
        def send():
            try:
                r = requests.post(url, json=data_dict, timeout=8)
                text = f"Status: {r.status_code}\n" + json.dumps(r.json(), ensure_ascii=False, indent=2)
            except Exception as e:
                text = f"请求失败：{e}"
            self.log_area.append(text)
            log_write(text)
            save_post_history(url, json.dumps(data_dict, ensure_ascii=False), text)
        Thread(target=send, daemon=True).start()
        self.log_area.append("[POST] 请求已发送…")

    def open_calc(self):
        CalcSignDialog(self).exec_()

    def open_read(self):
        ReadCodeDialog(self).exec_()

    def open_patch(self):
        BinPatchDialog(self).exec_()

if __name__ == "__main__":
    if sys.platform == "win32" and not is_admin():
        QMessageBox.information(None, "权限提示", "首次运行将申请管理员权限。")
        elevate()
        sys.exit()

    app = QApplication(sys.argv)
    app.setApplicationName("POSC工具箱")
    win = MainWindow()
    win.show()
    sys.exit(app.exec_())