岚图汽车
本文最后更新于58 天前,其中的信息可能已经过时,如有错误请5Yqg5b6u5L+ha2Fud2s2NjY=

因毕业找工作遇到的一个技术面试题但由于紧张等因素导致未能通过不过加密还是比较常规的,接下来是本次逆向的网站https://www.voyah.com.cn/store.html,实现的目标为采集全部门店类型,名称。

通过点击全国的下拉列表我们可以发现发出了getCatlist和findAllStore包 直接复制该curl发现返回{“code”:-10001,”msg”:”请求已失效”},发现请求参数里面有一个sign_encrypt可能与签名有关,同时也发现了请求负载中有req参数,对应的值为密文,故需要逆向分析

通过前端 JS 代码分析,定位到门店接口如下:https://clientapi.dsmp.voyah.com.cn/mygateway/v1/test/drive/app/queryStoreCityList

该接口由前端方法调用:

getStoreList: e => Bs({ url: "/mygateway/v1/test/drive/app/queryStoreCityList", method: "get", params: e })

因此,该接口为 GET 请求,业务参数通过 query params 传递。

加密与签名逻辑分析:

前端代码中存在 AES 加解密工具:

const bs = { encrypt(e, t) { const a = Da.enc.Utf8.parse(t) , s = Da.enc.Utf8.parse(e); return Da.AES.encrypt(s, a, { mode: Da.mode.ECB, padding: Da.pad.Pkcs7 }).toString() }, decrypt(e, t) { const a = Da.enc.Utf8.parse(t) , s = Da.AES.decrypt(e, a, { mode: Da.mode.ECB, padding: Da.pad.Pkcs7 }); return Da.enc.Utf8.stringify(s).toString() } }

由此可确定:

加密算法:AES
加密模式:ECB
填充方式:PKCS7
密文格式:Base64
IV:无
AES Key:LyfKMyl9SAGVslvB

此外,请求拦截器中还会为业务接口添加请求头:

e.headers.sign_encrypt = s
e.headers.platform = Vs

其中:

platform = websitePc
sign_encrypt = RSA 公钥加密 signKey
ign_encrypt 不是 AES 加密结果,而是 RSA 公钥加密结果。

完整流程如下:

1. 使用 RSA 公钥加密固定 AES Key:LyfKMyl9SAGVslvB
2. 请求 /mygateway/v2/sign/get 获取临时 signKey
3. 使用同一个 RSA 公钥加密 signKey
4. 将加密结果放入请求头 sign_encrypt
5. 同时设置请求头 platform=websitePc

获取 signKey 的接口为:

POST https://clientapi.dsmp.voyah.com.cn/mygateway/v2/sign/get

请求体为:

{
  "aesKey": "RSA加密后的固定AES Key",
  "platform": "websitePc"
}

返回结果中:

{
  "data": {
    "sign": "临时signKey"
  }
}

业务接口请求头最终包含:

platform: websitePc
sign_encrypt: RSA加密后的signKey

具体的python代码实现如下:

import json
import time
import base64
import requests
import pandas as pd
from Crypto.PublicKey import RSA
from Crypto.Cipher import PKCS1_v1_5
from Crypto.Cipher import AES
from Crypto.Util.Padding import unpad


BASE_URL = "https://clientapi.dsmp.voyah.com.cn"
PRIVATE_KEY = "LyfKMyl9SAGVslvB"
PLATFORM = "websitePc"

PUBLIC_KEY = (
    "MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAhn/fqBt9dB9uLrOQnDGjcLbtNL4sgui/920Im0Ivr37pqW8TURIV3ev//6Da3HyysjoYWSX0HOdJaPOW7Q1bG/jIeTucTcUHJr7/SCa88x8UNq8VM1goOJg7izNq66fNMvu88IiMU2SVX1LzaMJjlYb/aGKiP52PCFoyg+7Bq0/up1rlVSx5/SdAiBZox94SSJJnqo1/MCnkgJioaRsZESODqfDp7sk+FsKi5vU8t+rO1mnuTkFzqR1kBy5YmXyKcvCiXkzqfy9dtgoaBp/qESm/IT55ZP7Kjk13t/RJC15rT2UMxcgTvd8mIQK26lagJf2UcM6zcISFc2PtVYZsjQIDAQAB"
)


STORE_TYPE_MAP = {
    1: "岚图空间",
    2: "交付服务中心",
    3: "旗舰店",
    4: "全功能用户中心",
    5: "快闪店",
    6: "钣喷授权中心",
    7: "虚拟店",
    "1": "岚图空间",
    "2": "交付服务中心",
    "3": "旗舰店",
    "4": "全功能用户中心",
    "5": "快闪店",
    "6": "钣喷授权中心",
    "7": "虚拟店",
}


def rsa_encrypt(text: str) -> str:
    public_key_text = PUBLIC_KEY.strip().replace("\n", "").replace("\r", "").replace(" ", "")

    if len(public_key_text) != 392:
        print("当前 PUBLIC_KEY 长度:", len(public_key_text))
        raise ValueError("PUBLIC_KEY 长度异常,正常应为 392,请检查是否复制完整")

    pem = (
        "-----BEGIN PUBLIC KEY-----\n"
        + "\n".join(public_key_text[i:i + 64] for i in range(0, len(public_key_text), 64))
        + "\n-----END PUBLIC KEY-----"
    )

    public_key = RSA.import_key(pem)
    cipher = PKCS1_v1_5.new(public_key)
    encrypted = cipher.encrypt(text.encode("utf-8"))

    return base64.b64encode(encrypted).decode("utf-8")


def aes_ecb_decrypt(cipher_text: str, key: str = PRIVATE_KEY) -> str:
    cipher = AES.new(key.encode("utf-8"), AES.MODE_ECB)
    decrypted = unpad(
        cipher.decrypt(base64.b64decode(cipher_text)),
        AES.block_size
    )
    return decrypted.decode("utf-8")


def get_sign_key() -> str:
    url = BASE_URL + "/mygateway/v2/sign/get"

    payload = {
        "aesKey": rsa_encrypt(PRIVATE_KEY),
        "platform": PLATFORM
    }

    headers = {
        "User-Agent": "Mozilla/5.0",
        "Content-Type": "application/json;charset=UTF-8",
        "Accept": "application/json, text/plain, */*",
    }

    resp = requests.post(url, headers=headers, json=payload, timeout=15)

    print("获取 signKey 状态码:", resp.status_code)
    print("获取 signKey 返回:", resp.text[:300])

    result = resp.json()
    sign_key = result.get("data", {}).get("sign")

    if not sign_key:
        raise RuntimeError("获取 signKey 失败,请检查接口返回")

    return sign_key


def make_headers() -> dict:
    sign_key = get_sign_key()
    sign_encrypt = rsa_encrypt(sign_key)

    return {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120.0 Safari/537.36",
        "Accept": "application/json, text/plain, */*",
        "Content-Type": "application/json;charset=UTF-8",
        "platform": PLATFORM,
        "sign_encrypt": sign_encrypt,
        "Referer": "https://www.voyah.com.cn/",
        "Origin": "https://www.voyah.com.cn",
    }


def request_store_api(params=None) -> dict:
    url = BASE_URL + "/mygateway/v1/test/drive/app/queryStoreCityList"

    headers = make_headers()

    resp = requests.get(
        url,
        headers=headers,
        params=params or {},
        timeout=20
    )

    print("门店接口状态码:", resp.status_code)
    print("请求 URL:", resp.url)
    print("门店接口原始返回:", resp.text[:500])

    result = resp.json()

    if "resp" in result:
        plain = aes_ecb_decrypt(result["resp"])
        return json.loads(plain)

    return result


def get_value(obj: dict, keys: list, default=""):
    for key in keys:
        value = obj.get(key)
        if value not in [None, ""]:
            return value
    return default


def normalize_store_type(value):
    if value in [None, ""]:
        return ""

    if value in STORE_TYPE_MAP:
        return STORE_TYPE_MAP[value]

    value_str = str(value)

    if value_str in STORE_TYPE_MAP:
        return STORE_TYPE_MAP[value_str]

    return value_str


def extract_store_rows(data):
    rows = []

    def walk(obj):
        if isinstance(obj, dict):
            store_name = get_value(obj, [
                "storeName",
                "shopName",
                "name",
                "dealerName",
                "serviceCenterName",
                "companyName"
            ])

            address = get_value(obj, [
                "address",
                "storeAddress",
                "addr",
                "detailAddress",
                "fullAddress"
            ])

            store_type_raw = get_value(obj, [
                "storeTypeName",
                "storeType",
                "typeName",
                "type",
                "shopType",
                "dealerType"
            ])

            store_type = normalize_store_type(store_type_raw)

            # 判断像不像门店数据:必须同时有名称和地址
            if store_name and address:
                rows.append({
                    "门店类型": store_type,
                    "门店名称": store_name,
                    "门店地址": address
                })

            for value in obj.values():
                walk(value)

        elif isinstance(obj, list):
            for item in obj:
                walk(item)

    walk(data)

    # 去重
    unique_rows = []
    seen = set()

    for row in rows:
        key = (
            row.get("门店类型", ""),
            row.get("门店名称", ""),
            row.get("门店地址", "")
        )

        if key not in seen:
            seen.add(key)
            unique_rows.append(row)

    return unique_rows


def save_result(rows):
    df = pd.DataFrame(rows, columns=["门店类型", "门店名称", "门店地址"])

    print("采集数量:", len(df))
    print(df.head(20))

    df.to_excel("岚图城市门店.xlsx", index=False)
    df.to_csv("岚图城市门店.csv", index=False, encoding="utf-8-sig")

    print("保存完成:岚图城市门店.xlsx / 岚图城市门店.csv")


if __name__ == "__main__":
    # 不传参数,获取全部或默认城市门店
    params = {}

    result = request_store_api(params)

    print("解密后返回结构预览:")
    print(json.dumps(result, ensure_ascii=False, indent=2)[:3000])

    rows = extract_store_rows(result)

    save_result(rows)
文末附加内容
上一篇
下一篇