从混淆 JavaScript 到图片批量下载：一次完整的 Web 数据链路逆向分析

decoder_aliases = {"_0x43ac"}

changed = True

while changed:
    changed = False

    for left, right in assignments:
        if right in decoder_aliases and left not in decoder_aliases:
            decoder_aliases.add(left)
            changed = True

_0xXXXXXX("0xYY")

b"168998"

import base64
import json

from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
from cryptography.hazmat.primitives.ciphers.aead import AESGCM


def decrypt_payload(
    encrypted_base64: str,
    post_id: str
):
    encrypted_bytes = base64.b64decode(
        encrypted_base64
    )

    salt = encrypted_bytes[:16]
    iv = encrypted_bytes[16:28]
    ciphertext = encrypted_bytes[28:]

    hkdf = HKDF(
        algorithm=hashes.SHA256(),
        length=32,
        salt=salt,
        info=b"",
    )

    aes_key = hkdf.derive(
        str(post_id).encode("utf-8")
    )

    aesgcm = AESGCM(aes_key)

    plaintext = aesgcm.decrypt(
        iv,
        ciphertext,
        None
    )

    text = plaintext.decode("utf-8")

    return json.loads(text)

import requests
from bs4 import BeautifulSoup
from urllib.parse import urljoin, urlparse


def get_image_base(article_url):
    response = requests.get(article_url)
    response.raise_for_status()

    soup = BeautifulSoup(
        response.text,
        "html.parser"
    )

    img = soup.select_one(
        "figure img"
    )

    src = img.get("src")

    full_src = urljoin(
        article_url,
        src
    )

    parsed = urlparse(
        full_src
    )

    base_path = (
        parsed.path
        .rsplit("/", 1)[0]
        + "/"
    )

    return (
        f"{parsed.scheme}://"
        f"{parsed.netloc}"
        f"{base_path}"
    )

import base64
import json
import sys
from pathlib import Path
from urllib.parse import urljoin, urlparse

import requests
from bs4 import BeautifulSoup
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
from cryptography.hazmat.primitives.ciphers.aead import AESGCM


BASE_URL = "https://kkmzt.com"


def make_session():
    session = requests.Session()

    session.headers.update({
        "User-Agent": (
            "Mozilla/5.0 "
            "(Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 "
            "(KHTML, like Gecko) "
            "Chrome/152.0.0.0 "
            "Safari/537.36"
        )
    })

    return session


def get_image_base(
    session,
    post_id
):
    article_url = (
        f"{BASE_URL}/photo/{post_id}"
    )

    response = session.get(
        article_url,
        timeout=10
    )

    response.raise_for_status()

    soup = BeautifulSoup(
        response.text,
        "html.parser"
    )

    img = soup.select_one(
        "figure img"
    )

    if not img:
        raise RuntimeError(
            "figure img not found"
        )

    src = urljoin(
        article_url,
        img.get("src")
    )

    parsed = urlparse(src)

    base_path = (
        parsed.path
        .rsplit("/", 1)[0]
        + "/"
    )

    image_base = (
        f"{parsed.scheme}://"
        f"{parsed.netloc}"
        f"{base_path}"
    )

    return image_base, article_url


def fetch_encrypted_data(
    session,
    post_id,
    article_url
):
    api_url = (
        f"{BASE_URL}/app/post/p"
        f"?id={post_id}"
    )

    response = session.get(
        api_url,
        headers={
            "X-Requested-With":
                "XMLHttpRequest",

            "Referer":
                article_url,
        },
        timeout=10,
    )

    response.raise_for_status()

    return response.json()["data"]


def decrypt_image_list(
    encrypted_base64,
    post_id
):
    encrypted_bytes = (
        base64.b64decode(
            encrypted_base64
        )
    )

    salt = encrypted_bytes[:16]

    iv = encrypted_bytes[
        16:28
    ]

    ciphertext = encrypted_bytes[
        28:
    ]

    hkdf = HKDF(
        algorithm=hashes.SHA256(),
        length=32,
        salt=salt,
        info=b"",
    )

    aes_key = hkdf.derive(
        str(post_id)
        .encode("utf-8")
    )

    plaintext = AESGCM(
        aes_key
    ).decrypt(
        iv,
        ciphertext,
        None
    )

    return json.loads(
        plaintext.decode("utf-8")
    )


def download_images(
    session,
    filenames,
    image_base,
    output_dir,
    article_url
):
    output_dir.mkdir(
        parents=True,
        exist_ok=True
    )

    for index, filename in enumerate(
        filenames,
        start=1
    ):
        image_url = urljoin(
            image_base,
            filename
        )

        output_path = (
            output_dir
            / f"{index:02d}_{filename}"
        )

        print(
            f"[{index:02d}] "
            f"{image_url}"
        )

        response = session.get(
            image_url,
            headers={
                "Referer":
                    article_url
            },
            timeout=20
        )

        response.raise_for_status()

        output_path.write_bytes(
            response.content
        )


def main():
    if len(sys.argv) < 2:
        print(
            "Usage: "
            "python download.py POST_ID"
        )

        sys.exit(1)

    post_id = sys.argv[1]

    session = make_session()

    image_base, article_url = (
        get_image_base(
            session,
            post_id
        )
    )

    print(
        "Image base:",
        image_base
    )

    encrypted_data = (
        fetch_encrypted_data(
            session,
            post_id,
            article_url
        )
    )

    filenames = (
        decrypt_image_list(
            encrypted_data,
            post_id
        )
    )

    print(
        "Images:",
        len(filenames)
    )

    download_images(
        session,
        filenames,
        image_base,
        Path("downloads") / post_id,
        article_url,
    )


if __name__ == "__main__":
    main()

image_base = (
    "https://f.meizitu.net/"
    "image/2026/03/"
)

import argparse
import base64
import json
import sys
from pathlib import Path
from urllib.parse import urljoin, urlparse

import requests
from bs4 import BeautifulSoup
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
from cryptography.hazmat.primitives.ciphers.aead import AESGCM


DEFAULT_BASE_URL = "https://kkmzt.com"


def make_session():
    session = requests.Session()
    session.headers.update({
        "User-Agent": (
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 (KHTML, like Gecko) "
            "Chrome/152.0.0.0 Safari/537.36"
        ),
        "Accept": "*/*",
    })
    return session


def fetch_article_image_base(session, base_url, post_id, timeout=10):
    article_url = base_url.rstrip("/") + f"/photo/{post_id}"
    print(f"[i] Fetch article page: {article_url}")

    response = session.get(article_url, timeout=timeout)
    response.raise_for_status()

    soup = BeautifulSoup(response.text, "html.parser")
    img = soup.select_one("figure img")

    if not img:
        raise RuntimeError("Could not find first <figure><img>.")

    src = img.get("src")
    if not src:
        raise RuntimeError("The first figure img does not have src.")

    full_src = urljoin(article_url, src)
    parsed = urlparse(full_src)
    base_path = parsed.path.rsplit("/", 1)[0] + "/"
    image_base = f"{parsed.scheme}://{parsed.netloc}{base_path}"

    print(f"[+] First image src : {full_src}")
    print(f"[+] Image base URL  : {image_base}")

    return image_base, article_url


def fetch_encrypted_data(session, base_url, post_id, article_url, timeout=10):
    api_url = base_url.rstrip("/") + f"/app/post/p?id={post_id}"
    print(f"[i] Fetch API: {api_url}")

    response = session.get(
        api_url,
        headers={
            "Accept": "application/json, text/plain, */*",
            "X-Requested-With": "XMLHttpRequest",
            "Referer": article_url,
        },
        timeout=timeout,
    )

    response.raise_for_status()
    payload = response.json()

    if not isinstance(payload, dict) or "data" not in payload:
        raise RuntimeError("API response does not contain data.")

    encrypted_base64 = payload["data"]

    if not isinstance(encrypted_base64, str):
        raise RuntimeError("response.data is not string.")

    return encrypted_base64


def decrypt_image_list(encrypted_base64, post_id):
    encrypted_bytes = base64.b64decode(
        encrypted_base64,
        validate=True,
    )

    if len(encrypted_bytes) < 29:
        raise ValueError("Encrypted payload is too short.")

    salt = encrypted_bytes[:16]
    iv = encrypted_bytes[16:28]
    ciphertext = encrypted_bytes[28:]

    print("\n===== Encryption layout =====")
    print(f"total      : {len(encrypted_bytes)} bytes")
    print(f"salt       : {len(salt)} bytes")
    print(f"iv         : {len(iv)} bytes")
    print(f"ciphertext : {len(ciphertext)} bytes")

    key_material = str(post_id).encode("utf-8")

    hkdf = HKDF(
        algorithm=hashes.SHA256(),
        length=32,
        salt=salt,
        info=b"",
    )

    aes_key = hkdf.derive(key_material)
    aesgcm = AESGCM(aes_key)

    plaintext = aesgcm.decrypt(
        iv,
        ciphertext,
        None,
    )

    data = json.loads(plaintext.decode("utf-8"))

    if not isinstance(data, list):
        raise RuntimeError("Decrypted JSON is not a list.")

    return [
        item.strip()
        for item in data
        if isinstance(item, str) and item.strip()
    ]


def download_images(
    session,
    filenames,
    image_base,
    output_dir,
    article_url,
    timeout=20,
):
    output_dir.mkdir(parents=True, exist_ok=True)

    manifest = []
    success = 0

    for index, filename in enumerate(filenames, start=1):
        image_url = urljoin(image_base, filename)
        local_name = f"{index:02d}_{filename}"
        local_path = output_dir / local_name

        print(f"[{index:02d}/{len(filenames):02d}] {image_url}")

        try:
            response = session.get(
                image_url,
                headers={"Referer": article_url},
                timeout=timeout,
                stream=True,
            )
            response.raise_for_status()

            with local_path.open("wb") as f:
                for chunk in response.iter_content(chunk_size=1024 * 128):
                    if chunk:
                        f.write(chunk)

            size = local_path.stat().st_size
            success += 1

            manifest.append({
                "index": index,
                "filename": filename,
                "url": image_url,
                "local_file": local_name,
                "size": size,
                "status": "ok",
            })

        except Exception as exc:
            manifest.append({
                "index": index,
                "filename": filename,
                "url": image_url,
                "local_file": local_name,
                "size": 0,
                "status": "failed",
                "error": str(exc),
            })

    manifest_path = output_dir / "manifest.json"
    manifest_path.write_text(
        json.dumps(manifest, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )

    print(f"\n[+] Downloaded: {success}/{len(filenames)}")
    print(f"[+] Folder: {output_dir}")
    print(f"[+] Manifest: {manifest_path}")


def main():
    parser = argparse.ArgumentParser(
        description="Fetch, decrypt and download post images."
    )

    parser.add_argument("post_id", help="Post ID")
    parser.add_argument("--base-url", default=DEFAULT_BASE_URL)
    parser.add_argument("--output", default=None)
    parser.add_argument("--timeout", type=float, default=10)

    args = parser.parse_args()

    output_dir = (
        Path(args.output)
        if args.output
        else Path("downloads") / str(args.post_id)
    )

    session = make_session()

    try:
        image_base, article_url = fetch_article_image_base(
            session,
            args.base_url,
            args.post_id,
            args.timeout,
        )

        encrypted_data = fetch_encrypted_data(
            session,
            args.base_url,
            args.post_id,
            article_url,
            args.timeout,
        )

        filenames = decrypt_image_list(
            encrypted_data,
            args.post_id,
        )

        print("\n===== Image URLs =====")
        for index, filename in enumerate(filenames, start=1):
            print(f"{index:02d}. {urljoin(image_base, filename)}")

        download_images(
            session,
            filenames,
            image_base,
            output_dir,
            article_url,
            timeout=max(args.timeout, 20),
        )

    except Exception as exc:
        print(f"\n[ERROR] {type(exc).__name__}: {exc}")
        sys.exit(1)


if __name__ == "__main__":
    main()