def rc4_crypt(data: bytes, key: bytes) -> bytes:
    s = list(range(256))
    j = 0
    for i in range(256):
        j = (j + s[i] + key[i % len(key)]) & 0xFF
        s[i], s[j] = s[j], s[i]

    out = bytearray()
    i = j = 0
    for b in data:
        i = (i + 1) & 0xFF
        j = (j + s[i]) & 0xFF
        s[i], s[j] = s[j], s[i]
        out.append(b ^ s[(s[i] + s[j]) & 0xFF])
    return bytes(out)

key = b"xxxxxxxxxxxx"加密密钥
plaintext = rc4_crypt(bytes.fromhex(cipher_hex), key).decode("utf-8")
print(plaintext)

山寨转本地 proxy 的完整实现

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# 本文件是带中文注释的版本，代码逻辑保持原样，用于说明本地 RC4 接口代理的处理流程。
from __future__ import annotations

import argparse
import datetime as _dt
import hashlib
import hmac
import http.client
import json
import os
import select
import secrets
import socket
import subprocess
import sys
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, parse_qsl, urlparse

# 默认运行参数：加密密钥、应用编号、示例卡密、机器码和目标程序路径。
DEFAULT_KEY = "RC4加密密钥"
DEFAULT_APP = "10017"
DEFAULT_KAMI = "CRACK_CARD"
DEFAULT_MCODE = "CRACK_MACHINE"
ANCHOR_HMAC_KEY = "xxxxxxxxx"
DEFAULT_EXE = r"D:\work\test22po夜猫矩阵助手V4.1.exe"
# 端口占用时按顺序尝试这些备用端口，0 表示让系统自动分配空闲端口。
FALLBACK_PORTS = (18080, 28080, 38080, 0)
# 这些逐跳请求头只对当前连接有效，转发到上游前需要过滤。
HOP_BY_HOP_HEADERS = {
    "connection",
    "proxy-connection",
    "keep-alive",
    "proxy-authenticate",
    "proxy-authorization",
    "te",
    "trailers",
    "transfer-encoding",
    "upgrade",
}


# 这些默认值来自本地构建中观察到的运行时资源初始化内容。
# 这里不能把资源配置做成空壳：任务启动进入工作线程前会先读取运行时资源。
# 字段结构故意保持宽松，既兼容从 data 取值，也兼容直接从顶层字段取值。
ASSET_RUNTIME_DEFAULTS = {
    "KS_COOKIE_MERGE_API_URL": "http://kshz.yixinye.cn/api.php",
    "KS_COOKIE_MERGE_API_KEY": "ECVBmGu3Hytv5oN7ZjIpykliongrfmdk99999",
    "KS_COOKIE_MERGE_ENABLED": "0",
    "KS_APP_CLIENT_KEY": "3c2cd3f3",
    "KS_APP_DEVICE_MOD": "Xiaomi(MI 9)",
    "KS_APP_VER": "12.3.20.36000",
    "KS_CLIENT_SIGN": "_PZtjwG_sU-5v4wHmjKpGzTfW6bA4rylQgjeS0Klb8l4oeMPBkg2Hw",
    "KS_YEMAO_PARSER_AUTH_MODE": "sign",
    "KS_YEMAO_PARSER_TIMEOUT": "10",
    "KS_YEMAO_PARSER_EMPTY_GOODS_RETRIES": "2",
    "KS_YEMAO_PARSER_EMPTY_GOODS_RETRY_DELAY_MS": "800",
    "GENERAL_VIDEO_PARSER_MIN_INTERVAL_MS": "1500",
    "GENERAL_VIDEO_PARSER_MAX_RETRIES": "10",
    "BAIMA_API": "",
    "BAIMA_AUTH_API": "",
    "KS_DELETE_AFTER_UPLOAD": "0",
    "KS_DISABLE_SSL_VERIFY": "0",
    "COOKIE_SYNC_URL": "",
    "COOKIE_SYNC_SECRET": "my_cookie_sync_secret_change_me",
}

# 日志脱敏字段名，只要字段名命中这些片段就隐藏真实值。
REDACT_FIELD_NAMES = ("cookie", "token", "authorization", "password", "secret", "kami", "key", "cap_ticket")


# 递归脱敏日志内容，避免敏感字段和超长文本直接落盘。
def redact_for_log(value, *, max_len: int = 180):
    if isinstance(value, dict):
        out = {}
        for k, v in value.items():
            kl = str(k).lower()
            if any(x in kl for x in REDACT_FIELD_NAMES):
                out[k] = f"<redacted:{len(str(v))}>"
            else:
                out[k] = redact_for_log(v, max_len=max_len)
        return out
    if isinstance(value, list):
        return [redact_for_log(x, max_len=max_len) for x in value[:8]] + ([f"<...{len(value)-8} more>"] if len(value) > 8 else [])
    if isinstance(value, (str, bytes)):
        s = value.decode("utf-8", errors="replace") if isinstance(value, bytes) else value
        return s if len(s) <= max_len else s[:max_len] + f"<...{len(s)-max_len} more>"
    return value


# 统一写入结构化日志；开启详细模式时同步打印到控制台。
def proxy_log(server, event: str, **fields) -> None:
    rec = {"ts": time.strftime("%Y-%m-%d %H:%M:%S"), "event": event}
    rec.update(redact_for_log(fields))
    line = json.dumps(rec, ensure_ascii=False, separators=(",", ":"))
    if getattr(server, "verbose", False):
        print("[log] " + line, flush=True)
    log_file = getattr(server, "log_file", "") or ""
    if log_file:
        try:
            os.makedirs(os.path.dirname(os.path.abspath(log_file)), exist_ok=True)
            with open(log_file, "a", encoding="utf-8") as fp:
                fp.write(line + "\n")
        except Exception as exc:
            if getattr(server, "verbose", False):
                print(f"[!] write log failed: {exc!r}", flush=True)


# 多线程服务类，允许地址复用，便于频繁重启调试。
class ReusableThreadingHTTPServer(ThreadingHTTPServer):
    allow_reuse_address = True
    daemon_threads = True


# RC4 加解密核心函数：同一个过程同时用于加密和解密。
def rc4_crypt(key: bytes, data: bytes) -> bytes:
    if not key:
        raise ValueError("RC4 密钥不能为空")
    s = list(range(256))
    j = 0
    for i in range(256):
        j = (j + s[i] + key[i % len(key)]) & 0xFF
        s[i], s[j] = s[j], s[i]
    i = j = 0
    out = bytearray()
    for b in data:
        i = (i + 1) & 0xFF
        j = (j + s[i]) & 0xFF
        s[i], s[j] = s[j], s[i]
        out.append(b ^ s[(s[i] + s[j]) & 0xFF])
    return bytes(out)


# 把明文先做 RC4，再转成十六进制字符串作为接口传输格式。
def rc4_hex_encrypt(text: str, key: str) -> str:
    return rc4_crypt(key.encode("utf-8"), text.encode("utf-8")).hex()


# 清理并解析十六进制密文，然后用 RC4 还原成文本。
def rc4_hex_decrypt(hex_text: str, key: str) -> str:
    cleaned = "".join(ch for ch in hex_text if ch in "0123456789abcdefABCDEF")
    return rc4_crypt(key.encode("utf-8"), bytes.fromhex(cleaned)).decode("utf-8", errors="replace")


# 解密 data 参数，并按查询字符串格式拆成字段字典。
def parse_encrypted_query_blob(hex_text: str, key: str) -> dict:
    plain = rc4_hex_decrypt(hex_text, key)
    q = parse_qs(plain, keep_blank_values=True)
    req = {k: (v[0] if len(v) == 1 else v) for k, v in q.items()}
    # 保留足够的原始请求证据，方便复现签名猜测，同时不影响普通字段读取。
    req["_plain_query"] = plain
    req["_pairs"] = parse_qsl(plain, keep_blank_values=True)
    return req


# 通用摘要工具：把多个片段用分隔符拼入哈希对象。
def digest_hex(algo: str, *parts) -> str:
    h = hashlib.new(algo)
    for part in parts:
        h.update(str(part).encode("utf-8"))
        h.update(b"|")
    return h.hexdigest()


# 通用 HMAC 工具：字典会先序列化为稳定的紧凑 JSON。
def hmac_hex(algo: str, key: str, payload) -> str:
    if not isinstance(payload, str):
        payload = json.dumps(payload, ensure_ascii=False, separators=(",", ":"), sort_keys=True)
    return hmac.new(key.encode("utf-8"), payload.encode("utf-8"), algo).hexdigest()


# 生成普通会话接口参与签名的原始文本。
def authcore_payload(obj: dict) -> str:
    # 对齐响应签名校验逻辑：按 code、msg、time 拼出待签名文本。
    # 这里依赖字典的表现形式和插入顺序，保持和目标校验路径一致。
    return f"{obj.get('code')}|{obj.get('msg')}|{obj.get('time')}"


# 使用锚定密钥计算普通会话接口的 HMAC 签名。
def authcore_sign(obj: dict) -> str:
    return hmac.new(ANCHOR_HMAC_KEY.encode("utf-8"), authcore_payload(obj).encode("utf-8"), hashlib.sha256).hexdigest()


# 给普通接口响应补齐 sign 和 resp_sig 字段。
def attach_authcore_signatures(obj: dict) -> dict:
    obj["sign"] = authcore_sign(obj)
    # resp_sig 当前不参与核心校验，但保留稳定的一百二十八位十六进制字段。
    obj["resp_sig"] = hmac_hex("sha512", ANCHOR_HMAC_KEY, obj)
    return obj


# 生成登录接口响应参与签名的原始文本。
def login_payload(obj: dict) -> str:
    msg_json = json.dumps(obj.get("msg"), ensure_ascii=False, separators=(",", ":"))
    return f"{obj.get('code')}|{msg_json}|{obj.get('time')}"


# 计算登录响应签名，消息体使用紧凑 JSON 表示。
def login_sign(obj: dict) -> str:
    # 对齐登录响应签名校验：密钥固定，内容为 code、紧凑消息体、time。
    return hmac.new(ANCHOR_HMAC_KEY.encode("utf-8"), login_payload(obj).encode("utf-8"), hashlib.sha256).hexdigest()


# 给登录响应补齐签名字段，key 参数保留给调用接口一致性。
def attach_login_signatures(obj: dict, key: str) -> dict:
    obj["sign"] = login_sign(obj)
    obj["resp_sig"] = hmac_hex("sha512", ANCHOR_HMAC_KEY, {"sign": obj["sign"], "msg": obj.get("msg"), "time": obj.get("time")})
    return obj


# 计算文本的 MD5 十六进制摘要。
def md5_text(text: str) -> str:
    return hashlib.md5(text.encode("utf-8")).hexdigest()


# 按请求字段顺序重建待签名文本，用于校验客户端提交的 sign。
def request_sign_payload_from_pairs(pairs, app_key: str) -> str:
    """卡密接口签名猜测：除 sign 字段外按提交顺序拼接，再追加应用密钥。"""
    return "&".join(f"{k}={v}" for k, v in pairs if k != "sign" and not k.startswith("_")) + f"&{app_key}"


# 从解密后的请求中计算本地预期的 MD5 签名。
def compute_request_md5_sign(req: dict, app_key: str) -> tuple[str, str]:
    pairs = req.get("_pairs") or [(k, v) for k, v in req.items() if not str(k).startswith("_")]
    payload = request_sign_payload_from_pairs(pairs, app_key)
    return payload, md5_text(payload)


# 尝试解密 value 参数，同时返回原文、明文和错误信息。
def decode_value_plain(query: dict, key: str) -> tuple[str, str, str]:
    raw = query.get("value", [""])[0] if query.get("value") else ""
    if not raw:
        return "", "", ""
    try:
        plain = rc4_hex_decrypt(raw, key)
        return raw, plain, ""
    except Exception as exc:
        return raw, "", repr(exc)


# 根据选择的模式计算响应里的 check 字段及调试用原文。
def compute_response_check(
    *,
    mode: str,
    now: int,
    key: str,
    value_plain: str = "",
    value_raw: str = "",
    kami: str = "",
    vip: str = "",
    token: str = "",
) -> tuple[str, str, str]:
    # 对齐登录响应中的 check 字段算法：通常为时间、锚定密钥、原始 value 的摘要。
    # 如果请求没有 value，原始 value 按空字符串处理。
    selected = mode
    if selected == "auto":
        selected = "time_anchor_value_raw" if value_raw else "time_anchor"

    if selected == "time_anchor":
        payload = f"{now}{ANCHOR_HMAC_KEY}"
    elif selected == "time_anchor_value_raw":
        payload = f"{now}{ANCHOR_HMAC_KEY}{value_raw}"
    elif selected == "time_key":
        payload = f"{now}{key}"
    elif selected == "time_key_value_plain":
        payload = f"{now}{key}{value_plain}"
    elif selected == "time_key_value_raw":
        payload = f"{now}{key}{value_raw}"
    elif selected == "legacy":
        payload = f"{kami}|{vip}|{token}|{now}"
    elif selected == "empty":
        payload = ""
    else:
        raise ValueError(f"未知 check 模式: {mode}")
    return md5_text(payload), payload, selected


# 把命令行传入的会员到期时间解析成 Unix 时间戳。
def parse_vip_until(text: str | None) -> int | None:
    if not text:
        return None
    s = str(text).strip()
    if not s:
        return None
    if s.isdigit():
        return int(s)
    s = s.replace("T", " ")
    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M"):
        try:
            return int(_dt.datetime.strptime(s, fmt).timestamp())
        except ValueError:
            pass
    try:
        d = _dt.datetime.strptime(s, "%Y-%m-%d")
        d = d.replace(hour=23, minute=59, second=59)
        return int(d.timestamp())
    except ValueError as exc:
        raise argparse.ArgumentTypeError("请使用 Unix 时间戳、YYYY-MM-DD 或 YYYY-MM-DD HH:MM:SS") from exc


# 构造登录成功响应对象，包括会员到期、令牌、能力票据和签名。
def build_login_object(
    *,
    key: str,
    app: str,
    kami: str,
    mcode: str,
    vip_days: int,
    vip_until: int | None = None,
    now: int | None = None,
    token: str | None = None,
    check_mode: str = "auto",
    value_plain: str = "",
    value_raw: str = "",
) -> dict:
    now = int(time.time()) if now is None else int(now)
    vip = str(int(vip_until) if vip_until else now + vip_days * 86400)
    token = token or digest_hex("sha256", "token", kami, mcode, app, now, secrets.token_hex(8))
    cap_seed = digest_hex("sha512", "cap", token, kami, mcode, app, now)
    cap_ticket = f"{cap_seed}:{app}|{mcode}||{now + 1200}"
    msg = {"kami": kami, "vip": vip, "token": token, "cap_ticket": cap_ticket}
    check, check_payload, selected_check_mode = compute_response_check(
        mode=check_mode,
        now=now,
        key=key,
        value_plain=value_plain,
        value_raw=value_raw,
        kami=kami,
        vip=vip,
        token=token,
    )
    obj = {"code": 200, "msg": msg, "time": now, "check": check}
    obj["_debug_check_payload"] = check_payload
    obj["_debug_check_mode"] = selected_check_mode
    return attach_login_signatures(obj, key)


# 构造令牌或机器码校验类接口的加密响应对象。
def build_validate_object(*, key: str, app: str, token: str, mcode: str, now: int | None = None) -> dict:
    now = int(time.time()) if now is None else int(now)
    obj = {
        "code": 200,
        "msg": "ok",
        "valid": True,
        "app": app,
        "token": token,
        "mcode": mcode,
        "time": now,
        "check": hashlib.md5(f"{token}|{mcode}|{now}".encode("utf-8")).hexdigest(),
    }
    return attach_authcore_signatures(obj)


# 把响应字典序列化成 JSON，再用 RC4 加密并返回字节串。
def encrypted_json_response(obj: dict, key: str) -> bytes:
    body = json.dumps(obj, ensure_ascii=False, separators=(",", ":"))
    return rc4_hex_encrypt(body, key).encode("ascii")



# 把响应字典序列化成未加密 JSON 字节串。
def plain_json_response(obj: dict) -> bytes:
    return json.dumps(obj, ensure_ascii=False, separators=(",", ":")).encode("utf-8")


# 构造本地会话令牌校验接口的普通 JSON 响应。
def build_validate_token_object(token: str = "", mcode: str = "", now: int | None = None) -> dict:
    # 构造会话令牌校验接口期望的响应体。
    # 响应签名仍然使用 code、msg、time 三段内容和锚定密钥。
    now = int(time.time()) if now is None else int(now)
    obj = {
        "code": 200,
        "msg": "ok",
        "valid": True,
        "token": token,
        "markcode": mcode,
        "time": now,
        "check": md5_text(f"{now}{ANCHOR_HMAC_KEY}"),
    }
    return attach_authcore_signatures(obj)


# 构造运行时资源配置响应，提供功能开关、密钥、配置和地址列表。
def build_asset_config_object(now: int | None = None) -> dict:
    now = int(time.time()) if now is None else int(now)
    ttl = 86400
    feature_payload = {
        "sig3_service": {"enabled": True, "urls": []},
        "app_api_st": {"enabled": True, "urls": []},
        "drama_names": {"enabled": True, "urls": []},
        "yemao_parser": {
            "enabled": True,
            "keys": {"api_key": "", "upstream_base_url": ""},
            "config": {
                "auth_mode": ASSET_RUNTIME_DEFAULTS.get("KS_YEMAO_PARSER_AUTH_MODE", "sign"),
                "timeout": ASSET_RUNTIME_DEFAULTS.get("KS_YEMAO_PARSER_TIMEOUT", "10"),
                "empty_goods_retries": ASSET_RUNTIME_DEFAULTS.get("KS_YEMAO_PARSER_EMPTY_GOODS_RETRIES", "2"),
                "empty_goods_retry_delay_ms": ASSET_RUNTIME_DEFAULTS.get("KS_YEMAO_PARSER_EMPTY_GOODS_RETRY_DELAY_MS", "800"),
            },
        },
        "general_video_parser": {
            "enabled": True,
            "keys": {"app_id": "", "app_secret": "", "upstream_base_url": ""},
            "config": {
                "min_interval_ms": ASSET_RUNTIME_DEFAULTS.get("GENERAL_VIDEO_PARSER_MIN_INTERVAL_MS", "1500"),
                "max_retries": ASSET_RUNTIME_DEFAULTS.get("GENERAL_VIDEO_PARSER_MAX_RETRIES", "10"),
            },
        },
        "goods_parser": {"enabled": True, "keys": {"api_key": "", "upstream_base_url": ""}, "config": {}},
        "drama_parser": {"enabled": True, "keys": {"api_key": "", "upstream_base_url": ""}, "config": {}},
    }
    keys = {name: data.get("keys", {}) for name, data in feature_payload.items() if isinstance(data, dict)}
    config = {name: data.get("config", {}) for name, data in feature_payload.items() if isinstance(data, dict)}
    urls = {"sig3_urls": [], "app_api_st_urls": [], "drama_names_urls": [], "sign_api_urls": []}
    assets = {
        "enabled": True,
        "ok": True,
        "app": DEFAULT_APP,
        "app_id": DEFAULT_APP,
        "client_version": "4.1",
        "min_version": "0.0",
        "ttl": ttl,
        "expires": now + ttl,
        "server_time": now,
        "defaults": dict(ASSET_RUNTIME_DEFAULTS),
        "features": feature_payload,
        "keys": keys,
        "config": config,
        "urls": urls,
        "sig3_urls": urls["sig3_urls"],
        "app_api_st_urls": urls["app_api_st_urls"],
        "drama_names_urls": urls["drama_names_urls"],
        "sign_api_urls": urls["sign_api_urls"],
    }
    obj = {
        "code": 200,
        "msg": "ok",
        "time": now,
        "ttl": ttl,
        "data": assets,
        "assets": assets,
        "runtime": assets,
        "defaults": assets["defaults"],
        "features": assets["features"],
        "keys": assets["keys"],
        "config": assets["config"],
        "urls": assets["urls"],
        "sig3_urls": assets["sig3_urls"],
        "app_api_st_urls": assets["app_api_st_urls"],
        "drama_names_urls": assets["drama_names_urls"],
        "sign_api_urls": assets["sign_api_urls"],
        "check": md5_text(f"{now}{ANCHOR_HMAC_KEY}"),
    }
    return attach_authcore_signatures(obj)


# 为未单独实现的应用接口构造通用成功响应。
def build_generic_app_api_object(path: str, now: int | None = None) -> dict:
    now = int(time.time()) if now is None else int(now)
    obj = {
        "code": 200,
        "msg": "ok",
        "ok": True,
        "path": path,
        "time": now,
        "check": md5_text(f"{now}{ANCHOR_HMAC_KEY}"),
        "data": {},
    }
    return attach_authcore_signatures(obj)


# 同时从查询参数和 JSON 请求体中提取令牌与机器码。
def read_context_fields(query: dict, json_body: dict | None = None) -> tuple[str, str]:
    json_body = json_body or {}
    token = str(json_body.get("token") or query.get("token", [""])[0] or "")
    mcode = str(
        json_body.get("markcode")
        or json_body.get("mcode")
        or query.get("markcode", [""])[0]
        or query.get("mcode", [""])[0]
        or ""
    )
    return token, mcode


# 合并多个查询字符串，保留同名字段的多个值。
def merge_qs(*query_texts: str) -> dict:
    merged: dict[str, list[str]] = {}
    for query_text in query_texts:
        if not query_text:
            continue
        for k, vals in parse_qs(query_text, keep_blank_values=True).items():
            merged.setdefault(k, []).extend(vals)
    return merged


# 判断当前请求是否是需要本地模拟处理的 api.php 接口。
def is_kmlogon_api_path(path: str, query: dict) -> bool:
    api = (query.get("api", [""])[0] or "").lower()
    return (path.endswith("/api.php") or path == "/api.php") and bool(api)


# 解析 Host 头里的主机和端口，兼容 IPv6 方括号写法。
def parse_host_port(host_value: str, default_port: int = 80) -> tuple[str, int]:
    host_value = (host_value or "").strip()
    if not host_value:
        raise ValueError("缺少 Host 头")
    if host_value.startswith("["):
        end = host_value.find("]")
        if end < 0:
            raise ValueError(f"bad IPv6 host: {host_value!r}")
        host = host_value[1:end]
        rest = host_value[end + 1 :]
        port = int(rest[1:]) if rest.startswith(":") else default_port
        return host, port
    if host_value.count(":") == 1:
        host, port_s = host_value.rsplit(":", 1)
        if port_s.isdigit():
            return host, int(port_s)
    return host_value, default_port


# 核心接口分发：解密请求、识别 api 类型、构造对应响应和证据。
def handle_api(query: dict, *, key: str, vip_days: int, vip_until: int | None = None, verbose: bool = False, check_mode: str = "auto"):
    api = (query.get("api", [""])[0] or "").lower()
    app = query.get("app", [DEFAULT_APP])[0] or DEFAULT_APP
    req = {}
    if query.get("data"):
        try:
            req = parse_encrypted_query_blob(query["data"][0], key)
        except Exception as exc:
            req = {"_decode_error": repr(exc)}
    kami = str(req.get("kami") or query.get("kami", [DEFAULT_KAMI])[0] or DEFAULT_KAMI)
    # markcode 表示当前机器绑定值，mcode 可能是旧字段或历史残留。
    mcode = str(req.get("markcode") or req.get("mcode") or query.get("mcode", [DEFAULT_MCODE])[0] or DEFAULT_MCODE)
    token = str(req.get("token") or query.get("token", [""])[0] or "")
    value_raw, value_plain, value_decode_error = decode_value_plain(query, key)
    req_sign_payload, req_sign_expected = ("", "")
    if req:
        req_sign_payload, req_sign_expected = compute_request_md5_sign(req, key)

    if api == "kmlogon":
        obj = build_login_object(
            key=key,
            app=app,
            kami=kami,
            mcode=mcode,
            vip_days=vip_days,
            vip_until=vip_until,
            check_mode=check_mode,
            value_plain=value_plain,
            value_raw=value_raw,
        )
    elif "valid" in api or "verify" in api or "token" in api:
        obj = build_validate_object(key=key, app=app, token=token or digest_hex("sha256", kami, mcode), mcode=mcode)
    elif "unbind" in api:
        obj = {"code": 200, "msg": "unbind ok", "time": int(time.time())}
        attach_authcore_signatures(obj)
    else:
        obj = {"code": 200, "msg": "ok", "api": api, "time": int(time.time())}
        attach_authcore_signatures(obj)

    response_check_debug = {
        "mode": obj.get("_debug_check_mode"),
        "payload": obj.get("_debug_check_payload"),
        "md5": obj.get("check"),
    }
    evidence = {
        "api": api,
        "app": app,
        "decoded_request": req,
        "value_raw": value_raw,
        "value_plain": value_plain,
        "value_decode_error": value_decode_error,
        "request_sign_md5_payload": req_sign_payload,
        "request_sign_md5_expected": req_sign_expected,
        "request_sign_matches": (str(req.get("sign") or "").lower() == req_sign_expected.lower()) if req_sign_expected else None,
        "response_login_hmac_payload": login_payload(obj) if api == "kmlogon" else None,
        "response_check_debug": response_check_debug,
        "response_obj": obj,
    }
    # 调试专用字段只留在本地证据里，不写入加密响应体。
    obj.pop("_debug_check_payload", None)
    obj.pop("_debug_check_mode", None)
    if verbose:
        print(json.dumps(evidence, ensure_ascii=False, indent=2), flush=True)
    return 200, encrypted_json_response(obj, key), evidence


# HTTP 请求处理器，负责本地模拟、上游转发和 CONNECT 隧道。
class KmlogonHandler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"
    server_version = "CTFKmlogonProxy/1.2"

    # 发送普通响应，并显式设置长度、缓存策略和连接关闭。
    def _send_plain(self, status: int, body: bytes, content_type: str = "text/plain; charset=ascii"):
        self.send_response(status)
        self.send_header("Content-Type", content_type)
        self.send_header("Cache-Control", "no-store")
        self.send_header("Connection", "close")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    # 按 Content-Length 读取请求体，未提供长度时返回空字节串。
    def _read_body(self) -> bytes:
        length = int(self.headers.get("Content-Length", "0") or 0)
        return self.rfile.read(length) if length else b""

    # 非本地模拟接口走普通 HTTP 转发，保持代理的透明性。
    def _forward_http(self, method: str, parsed, path: str, query_text: str, host: str, body: bytes):
        if not getattr(self.server, "forward_other", True):
            return self._send_plain(502, b"CTF proxy only handles /api.php")

        target_host, target_port = parse_host_port(host, 80)
        upstream_path = path or "/"
        if query_text:
            upstream_path += "?" + query_text

        headers = {}
        for k, v in self.headers.items():
            lk = k.lower()
            if lk in HOP_BY_HOP_HEADERS:
                continue
            if lk == "host":
                headers[k] = host
            else:
                headers[k] = v
        headers["Connection"] = "close"
        if body and "Content-Length" not in headers:
            headers["Content-Length"] = str(len(body))

        try:
            conn = http.client.HTTPConnection(target_host, target_port, timeout=self.server.upstream_timeout)
            conn.request(method, upstream_path, body=body if body else None, headers=headers)
            resp = conn.getresponse()
            resp_body = resp.read()
            self.send_response(resp.status, resp.reason)
            for k, v in resp.getheaders():
                if k.lower() in HOP_BY_HOP_HEADERS or k.lower() == "content-length":
                    continue
                self.send_header(k, v)
            self.send_header("Connection", "close")
            self.send_header("Content-Length", str(len(resp_body)))
            self.end_headers()
            self.wfile.write(resp_body)
            proxy_log(self.server, "forward_success", url=f"http://{host}{upstream_path}", status=resp.status, bytes=len(resp_body))
            if self.server.verbose:
                print(f"{self.client_address[0]} -> pass http://{host}{upstream_path} status={resp.status}", flush=True)
        except Exception as exc:
            msg = f"upstream forward failed for http://{host}{upstream_path}: {exc!r}".encode("utf-8", errors="replace")
            proxy_log(self.server, "forward_failed", url=f"http://{host}{upstream_path}", error=repr(exc))
            if self.server.verbose:
                print("[!] " + msg.decode("utf-8", errors="replace"), flush=True)
            self._send_plain(502, msg, "text/plain; charset=utf-8")
        finally:
            try:
                conn.close()
            except Exception:
                pass

    # 统一处理 GET 和 POST：解析 URL、请求体、接口类型并返回结果。
    def _handle_method(self, method: str):
        body = self._read_body()
        parsed = urlparse(self.path)
        if parsed.scheme and parsed.netloc:
            path, query_text, host = parsed.path, parsed.query, parsed.netloc
        else:
            path, query_text, host = parsed.path, parsed.query, self.headers.get("Host", "")
        body_query_text = ""
        json_body = {}
        ctype = (self.headers.get("Content-Type") or "").split(";", 1)[0].strip().lower()
        if body and ctype == "application/x-www-form-urlencoded":
            body_query_text = body.decode("utf-8", errors="replace")
        elif body and ctype == "application/json":
            try:
                json_body = json.loads(body.decode("utf-8", errors="replace"))
            except Exception as exc:
                json_body = {"_json_decode_error": repr(exc)}
        query = merge_qs(query_text, body_query_text)
        proxy_log(self.server, "request", method=method, host=host, path=path, query_keys=sorted(query.keys()), content_type=ctype, body_len=len(body))

        if is_kmlogon_api_path(path, query):
            status, body, _ = handle_api(query, key=self.server.rc4_key, vip_days=self.server.vip_days, vip_until=self.server.vip_until, verbose=self.server.verbose, check_mode=self.server.check_mode)
            proxy_log(self.server, "mock_kmlogon", status=status, api=query.get("api", [""])[0])
            self._send_plain(status, body)
        elif getattr(self.server, "mock_session_apis", True) and path.endswith("/api/api/validate_token.php"):
            token, mcode = read_context_fields(query, json_body)
            obj = build_validate_token_object(token=token, mcode=mcode)
            proxy_log(self.server, "mock_validate_token", token_prefix=token[:12], markcode_prefix=mcode[:12], response=obj)
            if self.server.verbose:
                print(json.dumps({"api": "validate_token", "token_prefix": token[:12], "markcode_prefix": mcode[:12], "response_obj": obj}, ensure_ascii=False, indent=2), flush=True)
            self._send_plain(200, plain_json_response(obj), "application/json; charset=utf-8")
        elif getattr(self.server, "mock_session_apis", True) and path.endswith("/api/api/asset_config.php"):
            obj = build_asset_config_object()
            proxy_log(self.server, "mock_asset_config", response=obj)
            if self.server.verbose:
                print(json.dumps({"api": "asset_config", "response_obj": obj}, ensure_ascii=False, indent=2), flush=True)
            self._send_plain(200, plain_json_response(obj), "application/json; charset=utf-8")
        elif getattr(self.server, "mock_session_apis", True) and path.startswith("/api/api/") and ("123.56.184.140" in host or not host):
            obj = build_generic_app_api_object(path)
            proxy_log(self.server, "mock_generic_app_api", path=path, response=obj)
            self._send_plain(200, plain_json_response(obj), "application/json; charset=utf-8")
        else:
            self._forward_http(method, parsed, path, query_text, host, body)
        if self.server.verbose:
            print(f"{self.client_address[0]} -> host={host} path={path}", flush=True)

    # 处理 GET 请求。
    def do_GET(self):
        self._handle_method("GET")

    # 处理 POST 请求。
    def do_POST(self):
        self._handle_method("POST")

    # 处理 HTTPS 代理隧道，把客户端和目标主机之间的数据双向搬运。
    def do_CONNECT(self):
        if not getattr(self.server, "forward_other", True):
            return self._send_plain(502, b"CONNECT disabled")
        host = self.path
        try:
            target_host, target_port = parse_host_port(host, 443)
            upstream = socket.create_connection((target_host, target_port), timeout=self.server.upstream_timeout)
            self.send_response(200, "Connection Established")
            self.send_header("Proxy-Agent", self.server_version)
            self.end_headers()
            if self.server.verbose:
                print(f"{self.client_address[0]} -> tunnel CONNECT {target_host}:{target_port}", flush=True)
            self.connection.setblocking(False)
            upstream.setblocking(False)
            sockets = [self.connection, upstream]
            deadline = time.time() + self.server.tunnel_timeout
            while sockets and time.time() < deadline:
                readable, _, errored = select.select(sockets, [], sockets, 1.0)
                if errored:
                    break
                for src in readable:
                    try:
                        data = src.recv(65536)
                    except BlockingIOError:
                        continue
                    except OSError:
                        data = b""
                    if not data:
                        sockets = []
                        break
                    dst = upstream if src is self.connection else self.connection
                    dst.sendall(data)
        except Exception as exc:
            msg = f"CONNECT failed for {host}: {exc!r}".encode("utf-8", errors="replace")
            if self.server.verbose:
                print("[!] " + msg.decode("utf-8", errors="replace"), flush=True)
            try:
                self._send_plain(502, msg, "text/plain; charset=utf-8")
            except Exception:
                pass
        finally:
            try:
                upstream.close()
            except Exception:
                pass

    # 禁用默认访问日志，避免和自定义结构化日志混在一起。
    def log_message(self, fmt, *args):
        return


# 创建监听服务；如果端口被占用，可按配置自动尝试备用端口。
def create_server(host: str, requested_port: int, *, strict_port: bool = False):
    ports = [requested_port]
    if not strict_port:
        ports.extend(p for p in FALLBACK_PORTS if p != requested_port)
    last_exc = None
    for port in ports:
        try:
            httpd = ReusableThreadingHTTPServer((host, port), KmlogonHandler)
            return httpd, httpd.server_address[1]
        except OSError as exc:
            last_exc = exc
            print(f"[!] bind {host}:{port} failed: {exc}", flush=True)
    raise last_exc or RuntimeError("端口绑定失败")


# 自检流程：构造一份加密登录请求，验证本地响应能被正确解密。
def self_test() -> int:
    key = DEFAULT_KEY
    plain_req = "kami=TEST_CARD&mcode=TEST_MACHINE&t=1784706351&sign=TEST"
    data_hex = rc4_hex_encrypt(plain_req, key)
    value_hex = rc4_hex_encrypt(key, key)
    q = parse_qs(f"api=kmlogon&app=10017&data={data_hex}&value={value_hex}")
    status, body, evidence = handle_api(q, key=key, vip_days=7, vip_until=None, check_mode="auto")
    assert status == 200
    obj = json.loads(rc4_hex_decrypt(body.decode("ascii"), key))
    assert obj["code"] == 200 and obj["msg"]["kami"] == "TEST_CARD"
    assert obj["msg"]["cap_ticket"].endswith("|TEST_MACHINE||" + str(obj["time"] + 1200))
    print(json.dumps({"request": evidence["decoded_request"], "response": obj}, ensure_ascii=False, indent=2))
    return 0


# 启动目标前备份旧会话文件，避免历史令牌影响测试结果。
def clear_session_file_for(exe_path: str):
    token_path = os.path.join(os.path.dirname(os.path.abspath(exe_path)), "session.token")
    if os.path.exists(token_path):
        backup = token_path + ".bak"
        try:
            if os.path.exists(backup):
                os.remove(backup)
            os.replace(token_path, backup)
            print(f"[+] moved stale session.token -> {backup}", flush=True)
        except OSError as exc:
            print(f"[!] failed to move session.token: {exc}", flush=True)
    else:
        print("[+] no stale session.token", flush=True)


# 启动目标程序，并通过环境变量把网络请求导向本地代理端口。
def launch_target(exe_path: str, port: int):
    exe_path = os.path.abspath(exe_path)
    env = os.environ.copy()
    proxy = f"http://127.0.0.1:{port}"
    env["HTTP_PROXY"] = proxy
    env["HTTPS_PROXY"] = proxy
    env["NO_PROXY"] = ""
    cwd = os.path.dirname(exe_path) or None
    print(f"[+] launching target: {exe_path}", flush=True)
    print(f"[+] injected env: HTTP_PROXY={proxy} HTTPS_PROXY={proxy} NO_PROXY=", flush=True)
    return subprocess.Popen([exe_path], cwd=cwd, env=env)


# 命令行入口：解析参数、启动代理、可选启动目标程序并进入服务循环。
def main() -> int:
    ap = argparse.ArgumentParser(description="本地卡密登录 RC4 代理服务")
    ap.add_argument("--key", default=DEFAULT_KEY)
    ap.add_argument("--host", default="127.0.0.1")
    ap.add_argument("--port", type=int, default=18080)
    ap.add_argument("--vip-days", type=int, default=30)
    ap.add_argument("--vip-until", type=parse_vip_until, default=None, help='指定会员到期时间，例如 "2059-12-31 23:59:59"')
    ap.add_argument("--verbose", action="store_true")
    ap.add_argument("--self-test", action="store_true")
    ap.add_argument("--strict-port", action="store_true", help="端口绑定失败时不自动切换备用端口")
    ap.add_argument("--no-forward-other", action="store_true", help="关闭非卡密接口流量的透明转发")
    ap.add_argument("--no-mock-session-apis", action="store_true", help="关闭本地会话校验和资源配置接口模拟")
    ap.add_argument("--log-file", default=os.path.join(os.path.dirname(os.path.abspath(__file__)), "ctf_kmlogon_proxy.log"), help="把请求和决策日志写到此处；留空则关闭文件日志")
    ap.add_argument("--upstream-timeout", type=float, default=20.0, help="普通上游 HTTP 请求超时时间")
    ap.add_argument("--tunnel-timeout", type=float, default=600.0, help="HTTPS 隧道空闲存活时间")
    ap.add_argument(
        "--check-mode",
        default="auto",
        choices=("auto", "time_anchor", "time_anchor_value_raw", "time_key", "time_key_value_plain", "time_key_value_raw", "legacy", "empty"),
        help="响应 check 公式：auto 使用时间、锚定密钥和原始 value；legacy 使用旧令牌组",
    )
    ap.add_argument("--launch", nargs="?", const=DEFAULT_EXE, help="启动目标程序，并把 HTTP_PROXY 指向实际监听端口")
    ap.add_argument("--clear-session", action="store_true", help="启动前把目标目录里的 session.token 改名备份")
    args = ap.parse_args()

    if args.self_test:
        return self_test()

    httpd, actual_port = create_server(args.host, args.port, strict_port=args.strict_port)
    httpd.rc4_key = args.key
    httpd.vip_days = args.vip_days
    httpd.vip_until = args.vip_until
    httpd.verbose = args.verbose
    httpd.check_mode = args.check_mode
    httpd.forward_other = not args.no_forward_other
    httpd.mock_session_apis = not args.no_mock_session_apis
    httpd.log_file = args.log_file
    httpd.upstream_timeout = args.upstream_timeout
    httpd.tunnel_timeout = args.tunnel_timeout
    proxy = f"http://127.0.0.1:{actual_port}"
    vip_desc = args.vip_until if args.vip_until else f"+{args.vip_days}d"
    print(f"[+] listening http://{args.host}:{actual_port} key_len={len(args.key)} vip={vip_desc} check_mode={args.check_mode} forward_other={httpd.forward_other} mock_session_apis={httpd.mock_session_apis}", flush=True)
    print(f"[+] set env: HTTP_PROXY={proxy}  HTTPS_PROXY={proxy}  NO_PROXY=", flush=True)
    if args.log_file:
        print(f"[+] log_file={args.log_file}", flush=True)

    child = None
    if args.launch:
        if args.clear_session:
            clear_session_file_for(args.launch)
        child = launch_target(args.launch, actual_port)
        print(f"[+] target_pid={child.pid}", flush=True)

    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        print("\n[+] stopping", flush=True)
    finally:
        httpd.server_close()
    return 0


# 作为脚本运行时执行主函数，并把返回值作为进程退出码。
if __name__ == "__main__":
    raise SystemExit(main())

这个代码的原理和功能

这个脚本本质上是一个本地 HTTP 代理 / 接口模拟服务，主要用于本地调试或 逆向破解 场景里的 RC4 加密接口流程。

核心流程：

1. 启动本地代理服务

   • 默认监听 127.0.0.1:18080
   • 如果端口被占用，会尝试 28080、38080 或系统随机端口。

2. 识别特定接口

   • 主要处理类似：

   /api.php?api=kmlogon

   的请求。

3. 解密请求参数

   • 请求里的 data 参数是 RC4 加密后的十六进制字符串。
   • 脚本用 DEFAULT_KEY 解密后，解析出 kami、mcode、markcode、token 等字段。

4. 构造本地响应

   • 登录接口会返回：
     • 卡密
     • 到期时间
     • token
     • cap_ticket
     • check
     • sign
     • resp_sig

5. 签名和校验

   • 使用 MD5、HMAC-SHA256、HMAC-SHA512 生成响应里的校验字段。
   • check_mode 支持多种计算方式，默认 auto。

6. 其他接口模拟

   • /api/api/validate_token.php
   • /api/api/asset_config.php
   • /api/api/...

   这些接口会返回本地构造的 JSON 成功响应。

7. 普通流量转发

   • 如果请求不是脚本要模拟的接口，就会走普通 HTTP 转发。
   • HTTPS 的 CONNECT 请求也会建立隧道转发。

8. 日志脱敏

   • 日志里遇到 cookie、token、password、secret、kami、key 等字段会自动隐藏，避免敏感信息明文落盘。

9. 自检功能

   • --self-test 会构造一条测试登录请求：
     • 加密请求
     • 调用接口处理逻辑
     • 解密响应
     • 验证返回结构是否正常

10. 可选启动目标程序

    • --launch 可以启动指定 exe。
    • 启动时会自动设置：

HTTP_PROXY=http://127.0.0.1:端口
HTTPS_PROXY=http://127.0.0.1:端口

让目标程序流量经过这个本地代理。