179 lines
8.1 KiB
Python
Executable File
179 lines
8.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""WEVIA Stealth Browser — bypasses CF, captcha, bot detection
|
|
Usage: python3 wevia-stealth.py --url URL --action [fetch|screenshot|login|solve_captcha]
|
|
"""
|
|
import asyncio, json, sys, os, re
|
|
|
|
SECRETS = "/etc/weval/secrets.env"
|
|
def load_secrets():
|
|
s = {}
|
|
for l in open(SECRETS):
|
|
l = l.strip()
|
|
if '=' in l and not l.startswith('#'):
|
|
k, v = l.split('=', 1); s[k.strip()] = v.strip().strip('"\'')
|
|
return s
|
|
|
|
def save_secret(key, value):
|
|
content = open(SECRETS).read()
|
|
if re.search(f'^{key}=', content, re.M):
|
|
content = re.sub(f'^{key}=.*$', f'{key}={value}', content, flags=re.M)
|
|
else:
|
|
content += f"\n{key}={value}"
|
|
open(SECRETS, 'w').write(content)
|
|
|
|
async def stealth_browser(url, action="fetch", extra=None):
|
|
from playwright.async_api import async_playwright
|
|
|
|
async with async_playwright() as p:
|
|
# Launch with stealth settings
|
|
browser = await p.chromium.launch(
|
|
headless=True,
|
|
args=[
|
|
"--no-sandbox","--disable-dev-shm-usage",
|
|
"--disable-blink-features=AutomationControlled",
|
|
"--disable-infobars",
|
|
"--window-size=1920,1080",
|
|
"--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
|
|
]
|
|
)
|
|
|
|
ctx = await browser.new_context(
|
|
viewport={"width":1920,"height":1080},
|
|
user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
|
|
locale="fr-FR",
|
|
timezone_id="Africa/Casablanca"
|
|
)
|
|
|
|
# Anti-detection scripts
|
|
await ctx.add_init_script("""
|
|
Object.defineProperty(navigator, 'webdriver', {get: () => undefined});
|
|
Object.defineProperty(navigator, 'plugins', {get: () => [1,2,3,4,5]});
|
|
Object.defineProperty(navigator, 'languages', {get: () => ['fr-FR','fr','en-US','en']});
|
|
window.chrome = {runtime: {}};
|
|
const originalQuery = window.navigator.permissions.query;
|
|
window.navigator.permissions.query = (params) => (
|
|
params.name === 'notifications' ? Promise.resolve({state: Notification.permission}) : originalQuery(params)
|
|
);
|
|
""")
|
|
|
|
page = await ctx.new_page()
|
|
result = {"url": url, "action": action}
|
|
|
|
try:
|
|
resp = await page.goto(url, timeout=30000, wait_until="domcontentloaded")
|
|
result["status"] = resp.status if resp else 0
|
|
await page.wait_for_timeout(3000)
|
|
|
|
content = await page.content()
|
|
|
|
# Detect Cloudflare challenge
|
|
if "challenge-platform" in content or "cf-browser-verification" in content or "Just a moment" in content:
|
|
result["cloudflare"] = True
|
|
print("CF challenge detected → waiting...")
|
|
await page.wait_for_timeout(8000)
|
|
content = await page.content()
|
|
if "challenge-platform" not in content:
|
|
result["cf_bypass"] = "success"
|
|
else:
|
|
# Try cloudscraper fallback
|
|
result["cf_bypass"] = "waiting_longer"
|
|
await page.wait_for_timeout(15000)
|
|
content = await page.content()
|
|
|
|
# Detect CAPTCHA
|
|
if "captcha" in content.lower() or "recaptcha" in content.lower() or "hcaptcha" in content.lower():
|
|
result["captcha_detected"] = True
|
|
secrets = load_secrets()
|
|
solver_key = secrets.get("CAPSOLVER_KEY") or secrets.get("TWOCAPTCHA_KEY","")
|
|
if solver_key:
|
|
result["captcha_solver"] = "available"
|
|
# Extract sitekey
|
|
sitekey_match = re.search(r'sitekey["\s:=]+([a-f0-9-]{30,})', content)
|
|
if sitekey_match:
|
|
result["captcha_sitekey"] = sitekey_match.group(1)
|
|
else:
|
|
result["captcha_solver"] = "no_key_configured"
|
|
|
|
if action == "fetch":
|
|
title = await page.title()
|
|
text = await page.evaluate("document.body?.innerText?.substring(0,500)||''")
|
|
result["title"] = title
|
|
result["text"] = text[:300]
|
|
|
|
elif action == "screenshot":
|
|
path = f"/var/www/html/api/exports/screenshot-{int(__import__('time').time())}.png"
|
|
await page.screenshot(path=path, full_page=False)
|
|
result["screenshot"] = path
|
|
|
|
elif action == "login":
|
|
# Auto-detect login form
|
|
forms = await page.query_selector_all('form')
|
|
if forms:
|
|
email_input = await page.query_selector('input[type="email"], input[name="email"], input[name="username"], input[id="email"]')
|
|
pwd_input = await page.query_selector('input[type="password"]')
|
|
if email_input and pwd_input and extra:
|
|
creds = json.loads(extra) if isinstance(extra, str) else extra
|
|
await email_input.fill(creds.get("email",""))
|
|
await pwd_input.fill(creds.get("password",""))
|
|
submit = await page.query_selector('button[type="submit"], input[type="submit"]')
|
|
if submit:
|
|
await submit.click()
|
|
await page.wait_for_timeout(5000)
|
|
result["login"] = "submitted"
|
|
result["post_url"] = page.url
|
|
# Extract cookies for future use
|
|
cookies = await ctx.cookies()
|
|
result["cookies_count"] = len(cookies)
|
|
else:
|
|
result["login"] = "no_fields_found"
|
|
|
|
elif action == "extract_token":
|
|
# Look for token/API key on the page
|
|
tokens = re.findall(r'(hf_[a-zA-Z0-9]{20,}|sk-[a-zA-Z0-9]{20,}|ghp_[a-zA-Z0-9]{20,}|EAA[a-zA-Z0-9]{50,})', content)
|
|
if tokens:
|
|
result["tokens_found"] = len(tokens)
|
|
result["first_token"] = tokens[0][:20] + "..."
|
|
else:
|
|
# Check input fields
|
|
inputs = await page.query_selector_all('input[readonly], code, .token-value, pre')
|
|
for inp in inputs:
|
|
val = await inp.text_content() or await inp.get_attribute("value") or ""
|
|
if re.match(r'(hf_|sk-|ghp_|EAA)', val):
|
|
result["token_in_field"] = val[:20] + "..."
|
|
break
|
|
|
|
except Exception as e:
|
|
result["error"] = str(e)[:100]
|
|
|
|
await browser.close()
|
|
|
|
return result
|
|
|
|
async def cf_scrape(url):
|
|
"""Cloudflare bypass via cloudscraper"""
|
|
try:
|
|
import cloudscraper
|
|
scraper = cloudscraper.create_scraper(browser={"browser":"chrome","platform":"windows","mobile":False})
|
|
r = scraper.get(url, timeout=15)
|
|
return {"status": r.status_code, "length": len(r.text), "title": re.search(r'<title>(.*?)</title>', r.text, re.I).group(1) if re.search(r'<title>', r.text, re.I) else ""}
|
|
except Exception as e:
|
|
return {"error": str(e)[:80]}
|
|
|
|
async def main():
|
|
import argparse
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--url", default="https://huggingface.co/settings/tokens")
|
|
parser.add_argument("--action", default="fetch", choices=["fetch","screenshot","login","extract_token","cf_scrape"])
|
|
parser.add_argument("--extra", default=None)
|
|
args = parser.parse_args()
|
|
|
|
if args.action == "cf_scrape":
|
|
result = await cf_scrape(args.url)
|
|
else:
|
|
result = await stealth_browser(args.url, args.action, args.extra)
|
|
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|