2026年如何使用Camoufox绕过反机器人措施：移动代理与Playwright指南

import asyncio
from camoufox.async_api import AsyncCamoufox

async def scrape_target(browser, target_url):
    page = await browser.new_page()
    try:
        await page.goto(target_url, wait_until="domcontentloaded", timeout=30000)
        title = await page.title()
        print(f"Successfully scraped: {title}")
        return await page.content()
    finally:
        await page.close()

async def main():
    urls = [
        "https://browserleaks.com/javascript",
        "https://browserleaks.com/canvas",
        "https://browserleaks.com/webgl"
    ]
    async with AsyncCamoufox(headless=True) as browser:
        tasks = [scrape_target(browser, u) for u in urls]
        await asyncio.gather(*tasks)

if __name__ == "__main__":
    asyncio.run(main())

import requests
from camoufox.sync_api import Camoufox

# PXM2 Dedicated Mobile Proxy credentials
PROXY_HOST = "gate.pxm2.io"
PROXY_PORT = 30001
PROXY_USER = "pxm2_user_demo"
PROXY_PASS = "secret_pass_123"
ROTATION_URL = "https://pxm2.io/api/v1/rotate/userproxy_1029/"

proxy_config = {
    "server": f"http://{PROXY_HOST}:{PROXY_PORT}",
    "username": PROXY_USER,
    "password": PROXY_PASS,
}

def rotate_mobile_ip():
    """Trigger cellular modem reconnect to cycle mobile IP in 5-10 seconds."""
    resp = requests.get(ROTATION_URL, timeout=15)
    if resp.status_code == 200:
        print("[PXM2] Cellular IP successfully rotated!")

# Launch Camoufox with automatic location synchronization
with Camoufox(
    proxy=proxy_config,
    geoip=True,           # Aligns timezone, locale, and WebRTC to proxy IP
    os="windows",         # Consistent OS fingerprint
    headless=True
) as browser:
    page = browser.new_page()
    page.goto("https://iphey.com")
    print("Security evaluation loaded under genuine cellular carrier ASN!")

import asyncio
import aiohttp
from camoufox.async_api import AsyncCamoufox

# Concurrency & Worker Limits: 50 concurrent browser sessions
MAX_CONCURRENT_SESSIONS = 50
semaphore = asyncio.Semaphore(MAX_CONCURRENT_SESSIONS)

# Pool of 50 dedicated PXM2 4G/5G mobile modems (separate ports / credentials)
PXM2_MOBILE_PROXIES = [
    f"http://user_demo:pass_demo@gate.pxm2.io:{30000 + i}"
    for i in range(50)
]

# Remote IP rotation webhook API
PXM2_ROTATE_API = "https://pxm2.io/api/v1/rotate/modem_{id}/"

async def rotate_modem(modem_id: int):
    """Trigger remote modem cellular reconnect to cycle carrier IP in ~5-10s."""
    url = PXM2_ROTATE_API.format(id=modem_id)
    async with aiohttp.ClientSession() as session:
        async with session.get(url, timeout=10) as resp:
            if resp.status == 200:
                print(f"[Modem {modem_id}] Fresh cellular CGNAT IP acquired.")

async def run_worker(worker_id: int, target_url: str):
    """Worker handling single scraping task within the 50-concurrency semaphore boundary."""
    modem_id = worker_id % len(PXM2_MOBILE_PROXIES)
    proxy_endpoint = PXM2_MOBILE_PROXIES[modem_id]

    async with semaphore:
        # Launch lightweight Camoufox instance tailored for 50+ concurrency:
        # 1. geoip=True aligns timezone/locale/WebRTC to carrier egress IP automatically
        # 2. block_images=True saves cellular proxy bandwidth and cuts memory by 65%
        # 3. humanize=True generates authentic bezier mouse trajectories across challenges
        async with AsyncCamoufox(
            headless=True,
            geoip=True,
            proxy={"server": proxy_endpoint},
            humanize=True,
            block_images=True,
        ) as browser:
            page = await browser.new_page()
            try:
                print(f"[Worker {worker_id}] Connecting via Modem {modem_id} to {target_url}...")
                response = await page.goto(target_url, wait_until="domcontentloaded", timeout=45000)

                # Check if Cloudflare Turnstile or DataDome challenge is present
                turnstile_frame = page.frame_locator("iframe[src*='challenges.cloudflare.com']")
                checkbox = turnstile_frame.locator("input[type='checkbox']")
                
                if await checkbox.count() > 0:
                    print(f"[Worker {worker_id}] Cloudflare Turnstile challenge detected. Executing humanized click...")
                    await checkbox.click()
                    await page.wait_for_timeout(3000)

                # Validate bypass and extract required payload
                title = await page.title()
                if "Just a moment" in title or "Attention Required" in title:
                    print(f"[Worker {worker_id}] WAF challenge flagged! Triggering IP rotation on Modem {modem_id}...")
                    await rotate_modem(modem_id)
                    raise RuntimeError("WAF challenge encountered; IP rotated.")

                print(f"[Worker {worker_id}] Succeeded! Page Title: {title}")
                data = await page.content()
                return data

            except Exception as e:
                print(f"[Worker {worker_id}] Failed with error: {e}")
                return None
            finally:
                # Critical: page.close() prevents socket and memory leaks across long runs
                await page.close()

async def main():
    target_tasks = [
        f"https://target-portal.com/catalog/item/{i}" 
        for i in range(200)
    ]
    
    print(f"Starting {len(target_tasks)} tasks across {MAX_CONCURRENT_SESSIONS} concurrent workers...")
    jobs = [run_worker(i, url) for i, url in enumerate(target_tasks)]
    results = await asyncio.gather(*jobs, return_exceptions=True)
    successful = [r for r in results if r and not isinstance(r, Exception)]
    print(f"Batch completed: {len(successful)}/{len(target_tasks)} tasks resolved successfully.")

if __name__ == "__main__":
    asyncio.run(main())

from camoufox.sync_api import Camoufox

with Camoufox(
    humanize=True,        # Enables curved bezier mouse trajectories
    headless=False
) as browser:
    page = browser.new_page()
    page.goto("https://nowsecure.nl")
    
    # Cursor moves along a realistic physical path
    page.click("button#verify-button")
    page.wait_for_timeout(3000)

from camoufox.sync_api import Camoufox

# Store session cookies and local storage on disk
PROFILE_DIR = "./camoufox_profile"

with Camoufox(
    persistent_context=True,
    user_data_dir=PROFILE_DIR,
    disable_coop=True,    # Required to click elements inside cross-origin iframes
    headless=False,
    window=(1366, 768)
) as browser:
    page = browser.new_page()
    page.goto("https://target-portal.com/login")
    
    # If a Turnstile challenge is present, wait and click checkbox
    page.wait_for_selector("iframe[src*='challenges.cloudflare.com']", timeout=10000)
    page.wait_for_timeout(2000)
    
    # Click within iframe boundary
    turnstile_box = page.frame_locator("iframe[src*='challenges.cloudflare.com']")
    turnstile_box.locator("input[type='checkbox']").click()
    page.wait_for_timeout(5000)
    
    # Next execution reuses the cf_clearance cookie from PROFILE_DIR!

from camoufox.sync_api import Camoufox

# Linux headless server requires xvfb package installed:
# sudo apt-get install -y xvfb libgtk-3-0 libasound2

with Camoufox(
    headless="virtual",   # Uses Xvfb virtual display buffer
    block_images=True,    # Optimizes proxy bandwidth usage
    block_webrtc=False    # Spoofs WebRTC IP to match proxy without leaking local IP
) as browser:
    page = browser.new_page()
    page.goto("https://bot.incolumitas.com")
    print("Passed Incolumitas bot detection test on headless server!")