Script

How to parse the golden apple? PART 2

Всем привет!

Это продолжение первой темы . Приглашаю вас сначала ознакомиться с ней, там есть весь исходный контекст.
Ниже расскажу что получилось сделать в первый заход, что сломалось и какие выводы я из этого сделал.
Приятного прочтения!

Что удалось сделать в первый раз:

После нашей первой битвы удалось написать парсер, который собирает наличие товаров (это самый сложный эндпоинт, единственный который имеет хорошую защиту) - парсер делал это при помощи того, что захватывал нужные для запроса токены при помощи отдельных запросов на event эндпоинт.

Пайплайн был такой:
1) Отправляем запрос на event эндпоинт (он без защиты)
2) Получаем часть токенов с event-эндпоинта и генерируем недостающие на их основе (код генерации ниже)
3) Отправляем запрос на stock-эндпоинт для получения наличия товаров

парсер спокойно отработал несколько дней.

Что пошло не так

Через несколько дней парсер превратился в тыкву, эндпоинт наличия товаров начал отдавать 403.

Два дня я разбирался что не так, при этом event-эндпоинт по-прежнему отвечает корректно, структура ответа не изменилась.
Однако сгенерированные на его основе токены перестали проходить валидацию на stock-эндпоинте.

Как выглядит запрос на stock-эндпоинт и какие у него необходимые параметры. (скопировал curl с браузера)


Токены:

На текущий момент мне кажется дело обстоит в этих токенах:
x-gib-gsscw-goldapple - его собираем из event запроса
x-gib-fgsscw-goldapple - его генерируем на основании первого токена
ngx_s_id - его собираем из event запроса

Инструкцию для генерации токена fgsscw брал из js файла, который участвует в инициации эндпоинта

Код, которым я получаю токены

Для удобства, прикрепляю код, которым я доставал токены из event-эндпоинта, чтобы если вдруг что могли использовать его как основу, он:
1) захватывает event-запрос из браузера
2) извлекает gssc
3) генерирует fgsscw локально

import time
import random
import string
import hashlib
import requests
from urllib.parse import urlparse
from playwright.sync_api import sync_playwright

from parser.logger import get_logger
logger = get_logger()

LAUNCH_ARGS = [
"--disable-blink-features=AutomationControlled",
"--disable-infobars",
"--lang=ru-RU",
"--window-size=1920,1080",
"--disable-features=IsolateOrigins,site-per-process",
"--disable-web-security",
"--allow-running-insecure-content",
"--no-sandbox",
"--disable-setuid-sandbox",
]

S = string.ascii_letters + string.digits
UA = "shgkla34ty3gg354g34wf"


def gen_fgssc(gssc: str) -> str:
i = "".join(random.choice(S) for _ in range(4))
payload = (UA + gssc + i).encode("utf-8")
sha1_hex = hashlib.sha1(payload).hexdigest()
return i + sha1_hex[4:]


def close_popups(page):
time.sleep(0.35)
popup_steps = [
("role", "да, верно"),
("role", "хорошо"),
("role", "понятно"),
("css", 'button[class*="ga-modal__close-button"], button[class*="close-button"], button[aria-label="Close"]'),
("css", '[data-testid="modal-close"]'),
]
for kind, value in popup_steps:
try:
if kind == "role":
loc = page.get_by_role("button", name=value, exact=False).first
else:
loc = page.locator(value).first
if loc.is_visible(timeout=1000):
loc.click(delay=100)
time.sleep(0.2)
except Exception:
continue


def _parse_proxy_for_playwright(proxy_url: str) -> dict | None:
if not proxy_url:
return None
u = urlparse(proxy_url)
if not u.scheme or not u.hostname or not u.port:
return None
cfg = {"server": f"{u.scheme}://{u.hostname}:{u.port}"}
if u.username:
cfg["username"] = u.username
if u.password:
cfg["password"] = u.password
return cfg


class TokenHarvester:
def __init__(self, start_url: str, proxy_url: str | None = None):
self.user_agent = (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/143.0.0.0 Safari/137.36"
)
self.start_url = start_url
self.proxy_url = proxy_url

self.event_url = None
self.event_headers = None
self.event_post_data = None
self.captured_cookies = []

self.session = requests.Session()
if proxy_url:
self.session.proxies = {"http": proxy_url, "https": proxy_url}

self.base_headers = {
"accept": "application/json, text/plain, */*",
"accept-language": "ru-RU,ru;q=0.9,en-US;q=0.8,en;q=0.7",
"cache-control": "no-cache",
"pragma": "no-cache",
"sec-ch-ua": '"Google Chrome";v="143", "Chromium";v="143", "Not A(Brand";v="24"',
"sec-ch-ua-mobile": "?0",
"sec-ch-ua-platform": '"Windows"',
"sec-fetch-dest": "empty",
"sec-fetch-mode": "cors",
"sec-fetch-site": "same-origin",
"user-agent": self.user_agent,
}

self._capture_event()
if not self.event_url:
raise RuntimeError("Не удалось захватить event-запрос")

self._set_initial_cookies()
self.refresh_tokens()

def _capture_event(self):
logger.info("[harvester] запуск браузера для захвата event-запроса...")
captured = {"ok": False}
pw_proxy = _parse_proxy_for_playwright(self.proxy_url)

with sync_playwright() as p:
browser = p.chromium.launch(
headless=False,
channel="chrome",
args=LAUNCH_ARGS,
)

context_kwargs = {
"viewport": {"width": 1920, "height": 1080},
"locale": "ru-RU",
"user_agent": self.user_agent,
"ignore_https_errors": True,
}
if pw_proxy:
context_kwargs["proxy"] = pw_proxy

context = browser.new_context(**context_kwargs)
page = context.new_page()

def handle_request(request):
if captured["ok"]:
return
if request.method == "POST" and "front/api/event" in request.url:
self.event_url = request.url
self.event_headers = request.headers
self.event_post_data = request.post_data
captured["ok"] = True
logger.success(f"[harvester] event-запрос захвачен: {self.event_url}")

context.on("request", handle_request)

page.goto(self.start_url, wait_until="domcontentloaded", timeout=60000)

deadline = time.time() + 30
while time.time() < deadline and not captured["ok"]:
close_popups(page)
page.evaluate("window.scrollBy(0, window.innerHeight)")
time.sleep(0.5)

self.captured_cookies = context.cookies()

context.close()
browser.close()

def _set_initial_cookies(self):
for c in self.captured_cookies:
self.session.cookies.set(
c["name"],
c["value"],
domain=c.get("domain"),
path=c.get("path"),
)

def refresh_tokens(self):
logger.info("[harvester] обновление токенов через event-запрос...")

resp = self.session.post(
self.event_url,
headers=self.event_headers,
data=self.event_post_data,
timeout=30,
)
resp.raise_for_status()

data = resp.json()
d = data["data"]

gssc = d.get("cs", {}).get("gssc")
cfids = d.get("cfids")

fgsscw = gen_fgssc(gssc)

self.session.cookies.set("gsscw-goldapple", gssc, domain=".goldapple.ru", path="/")
self.session.cookies.set("fgsscw-goldapple", fgsscw, domain=".goldapple.ru", path="/")
if cfids:
self.session.cookies.set("cfidsw-goldapple", cfids, domain=".goldapple.ru", path="/")

self.base_headers["x-gib-gsscw-goldapple"] = gssc
self.base_headers["x-gib-fgsscw-goldapple"] = fgsscw

logger.success("[harvester] токены обновлены")

def get_session_and_headers(self):
return self.session, self.base_headers.copy()


def pick_cookie(jar, name: str, prefer_domain: str | None = ".goldapple.ru") -> str | None:
matches = [c for c in jar if c.name == name]
if not matches:
return None

if prefer_domain:
preferred = [c for c in matches if (c.domain or "").endswith(prefer_domain)]
if preferred:
preferred.sort(key=lambda c: len(c.path or "/"))
return preferred[0].value

matches.sort(key=lambda c: len(c.path or "/"))
return matches[0].value


def main():
h = TokenHarvester(
start_url="https://goldapple.ru/parfjumerija",
proxy_url=None,
)

session, headers = h.get_session_and_headers()

tokens = {
"gsscw-goldapple": pick_cookie(session.cookies, "gsscw-goldapple"),
"fgsscw-goldapple": pick_cookie(session.cookies, "fgsscw-goldapple"),
"cfidsw-goldapple": pick_cookie(session.cookies, "cfidsw-goldapple"),
"x-gib-gsscw-goldapple": headers.get("x-gib-gsscw-goldapple"),
"x-gib-fgsscw-goldapple": headers.get("x-gib-fgsscw-goldapple"),
}

print(tokens)


if __name__ == "__main__":
main()

Python

import time

import random

import string

import hashlib

import requests

from urllib.parse import urlparse

from playwright.sync_api import sync_playwright



from parser.logger import get_logger

logger = get_logger()



LAUNCH_ARGS = [

    "--disable-blink-features=AutomationControlled",

    "--disable-infobars",

    "--lang=ru-RU",

    "--window-size=1920,1080",

    "--disable-features=IsolateOrigins,site-per-process",

    "--disable-web-security",

    "--allow-running-insecure-content",

    "--no-sandbox",

    "--disable-setuid-sandbox",

]



S = string.ascii_letters + string.digits

UA = "shgkla34ty3gg354g34wf"





def gen_fgssc(gssc: str) -> str:

    i = "".join(random.choice(S) for _ in range(4))

    payload = (UA + gssc + i).encode("utf-8")

    sha1_hex = hashlib.sha1(payload).hexdigest()

    return i + sha1_hex[4:]





def close_popups(page):

    time.sleep(0.35)

    popup_steps = [

        ("role", "да, верно"),

        ("role", "хорошо"),

        ("role", "понятно"),

        ("css", 'button[class*="ga-modal__close-button"], button[class*="close-button"], button[aria-label="Close"]'),

        ("css", '[data-testid="modal-close"]'),

    ]

    for kind, value in popup_steps:

        try:

            if kind == "role":

                loc = page.get_by_role("button", name=value, exact=False).first

            else:

                loc = page.locator(value).first

            if loc.is_visible(timeout=1000):

                loc.click(delay=100)

                time.sleep(0.2)

        except Exception:

            continue





def _parse_proxy_for_playwright(proxy_url: str) -> dict | None:

    if not proxy_url:

        return None

    u = urlparse(proxy_url)

    if not u.scheme or not u.hostname or not u.port:

        return None

    cfg = {"server": f"{u.scheme}://{u.hostname}:{u.port}"}

    if u.username:

        cfg["username"] = u.username

    if u.password:

        cfg["password"] = u.password

    return cfg





class TokenHarvester:

    def __init__(self, start_url: str, proxy_url: str | None = None):

        self.user_agent = (

            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "

            "(KHTML, like Gecko) Chrome/143.0.0.0 Safari/137.36"

        )

        self.start_url = start_url

        self.proxy_url = proxy_url



        self.event_url = None

        self.event_headers = None

        self.event_post_data = None

        self.captured_cookies = []



        self.session = requests.Session()

        if proxy_url:

            self.session.proxies = {"http": proxy_url, "https": proxy_url}



        self.base_headers = {

            "accept": "application/json, text/plain, */*",

            "accept-language": "ru-RU,ru;q=0.9,en-US;q=0.8,en;q=0.7",

            "cache-control": "no-cache",

            "pragma": "no-cache",

            "sec-ch-ua": '"Google Chrome";v="143", "Chromium";v="143", "Not A(Brand";v="24"',

            "sec-ch-ua-mobile": "?0",

            "sec-ch-ua-platform": '"Windows"',

            "sec-fetch-dest": "empty",

            "sec-fetch-mode": "cors",

            "sec-fetch-site": "same-origin",

            "user-agent": self.user_agent,

        }



        self._capture_event()

        if not self.event_url:

            raise RuntimeError("Не удалось захватить event-запрос")



        self._set_initial_cookies()

        self.refresh_tokens()



    def _capture_event(self):

        logger.info("[harvester] запуск браузера для захвата event-запроса...")

        captured = {"ok": False}

        pw_proxy = _parse_proxy_for_playwright(self.proxy_url)



        with sync_playwright() as p:

            browser = p.chromium.launch(

                headless=False,

                channel="chrome",

                args=LAUNCH_ARGS,

            )



            context_kwargs = {

                "viewport": {"width": 1920, "height": 1080},

                "locale": "ru-RU",

                "user_agent": self.user_agent,

                "ignore_https_errors": True,

            }

            if pw_proxy:

                context_kwargs["proxy"] = pw_proxy



            context = browser.new_context(**context_kwargs)

            page = context.new_page()



            def handle_request(request):

                if captured["ok"]:

                    return

                if request.method == "POST" and "front/api/event" in request.url:

                    self.event_url = request.url

                    self.event_headers = request.headers

                    self.event_post_data = request.post_data

                    captured["ok"] = True

                    logger.success(f"[harvester] event-запрос захвачен: {self.event_url}")



            context.on("request", handle_request)



            page.goto(self.start_url, wait_until="domcontentloaded", timeout=60000)



            deadline = time.time() + 30

            while time.time() < deadline and not captured["ok"]:

                close_popups(page)

                page.evaluate("window.scrollBy(0, window.innerHeight)")

                time.sleep(0.5)



            self.captured_cookies = context.cookies()



            context.close()

            browser.close()



    def _set_initial_cookies(self):

        for c in self.captured_cookies:

            self.session.cookies.set(

                c["name"],

                c["value"],

                domain=c.get("domain"),

                path=c.get("path"),

            )



    def refresh_tokens(self):

        logger.info("[harvester] обновление токенов через event-запрос...")



        resp = self.session.post(

            self.event_url,

            headers=self.event_headers,

            data=self.event_post_data,

            timeout=30,

        )

        resp.raise_for_status()



        data = resp.json()

        d = data["data"]



        gssc = d.get("cs", {}).get("gssc")

        cfids = d.get("cfids")



        fgsscw = gen_fgssc(gssc)



        self.session.cookies.set("gsscw-goldapple", gssc, domain=".goldapple.ru", path="/")

        self.session.cookies.set("fgsscw-goldapple", fgsscw, domain=".goldapple.ru", path="/")

        if cfids:

            self.session.cookies.set("cfidsw-goldapple", cfids, domain=".goldapple.ru", path="/")



        self.base_headers["x-gib-gsscw-goldapple"] = gssc

        self.base_headers["x-gib-fgsscw-goldapple"] = fgsscw



        logger.success("[harvester] токены обновлены")



    def get_session_and_headers(self):

        return self.session, self.base_headers.copy()





def pick_cookie(jar, name: str, prefer_domain: str | None = ".goldapple.ru") -> str | None:

    matches = [c for c in jar if c.name == name]

    if not matches:

        return None



    if prefer_domain:

        preferred = [c for c in matches if (c.domain or "").endswith(prefer_domain)]

        if preferred:

            preferred.sort(key=lambda c: len(c.path or "/"))

            return preferred[0].value



    matches.sort(key=lambda c: len(c.path or "/"))

    return matches[0].value





def main():

    h = TokenHarvester(

        start_url="https://goldapple.ru/parfjumerija",

        proxy_url=None,

    )



    session, headers = h.get_session_and_headers()



    tokens = {

        "gsscw-goldapple": pick_cookie(session.cookies, "gsscw-goldapple"),

        "fgsscw-goldapple": pick_cookie(session.cookies, "fgsscw-goldapple"),

        "cfidsw-goldapple": pick_cookie(session.cookies, "cfidsw-goldapple"),

        "x-gib-gsscw-goldapple": headers.get("x-gib-gsscw-goldapple"),

        "x-gib-fgsscw-goldapple": headers.get("x-gib-fgsscw-goldapple"),

    }



    print(tokens)





if __name__ == "__main__":

    main()

Раньше схема была простой:
ngx_s_id мы просто брали из cookies event-запросов, ловили их в браузере, тестировали curl в Postman, убирали всё лишнее и stock-эндпоинт стабильно отвечал 200.

Сейчас этот подход перестал работать.
При этом:
1) event по-прежнему отвечает нормально
2) gsscw / fgsscw выглядят валидными
3) но stock стабильно возвращает 403

По ощущениям, проблема именно в ngx-части:
1) либо ngx_s_id теперь жёстко привязан к браузерному контексту
2) либо сервер проверяет связку ngx_s_id + fgsscw
3) либо изменился алгоритм генерации fgsscw
4) либо всё сразу

Цель простая и конкретная:

Понять, какой именно токен или связка токенов теперь ломает запрос и получить минимальный воспроизводимый пример, где токены получены корректно, stock-эндпоинт возвращает 200 (не целый парсер, а изолированное решение этого вопроса в коде)

Про вознаграждение и формат участия:

Как и в первый раз, проект остается некоммерческим, я делаю его для себя и из интереса к разбору защиты.
Но я уважаю чужое время, тем более задача на этот раз более углубленная.

Для человека, который реально разберётся, что именно изменилось и покажет рабочий минимальный пример, я готов поблагодарить автора переводом на счёт lolz. Если решение найдётся - я свяжусь с автором, переведу награду и приложу скриншот.

Важно:
пожалуйста, не пишите в личные сообщения с предложениями услуг.
Хочется, чтобы обсуждение и решение остались в рамках темы, так результат будет полезен всем а не только мне.

Заранее благодарю всех, кто прочитал и всех, кто приложит к этому делу руку!