"""reCAPTCHA v2 audio-challenge bypass using a local Whisper ASR model. The audio challenge is the accessibility mode of reCAPTCHA v2. Google offers it as a legitimate alternative to the image grid for visually-impaired users; the audio clip contains spoken digits and we transcribe them with Whisper. This approach is: * local (no 2Captcha API key, no per-solve cost) * fast (~1 s/solve with the `tiny` model on CPU) * high-accuracy (~85-95% on reCAPTCHA audio per IEEE EuroS&PW 2025) The solver operates on a Playwright Page that's already rendered the page with the captcha embedded. It drives the challenge iframe directly. Flow: 1. Locate the anchor iframe, click the checkbox to trigger the challenge 2. Locate the bframe (challenge iframe), click the audio button 3. Grab the audio `src`, download locally 4. Transcribe with Whisper → digits 5. Type into the response field, click verify 6. Read the g-recaptcha-response textarea to confirm token was issued """ import asyncio import logging import os import tempfile from pathlib import Path from typing import Any import httpx logger = logging.getLogger(__name__) # Module-level singleton so we load the 40 MB Whisper model exactly once per # worker. Lazy — imported in-function so scheduler code not using captchas # doesn't pay the import cost. _WHISPER_MODEL: Any = None def _get_whisper_model(): """Load Whisper on first use; reuse on every subsequent solve.""" global _WHISPER_MODEL if _WHISPER_MODEL is None: from faster_whisper import WhisperModel model_name = os.environ.get("WHISPER_MODEL", "tiny") cache = os.environ.get("HF_HOME", "/app/hf_cache") _WHISPER_MODEL = WhisperModel( model_name, device="cpu", compute_type="int8", download_root=cache, ) logger.info("whisper model loaded: %s (int8)", model_name) return _WHISPER_MODEL async def _transcribe_audio(audio_bytes: bytes) -> str: """Run Whisper on raw MP3 bytes. Returns the best text transcription (lowercase, trimmed). Does the decode+infer call in a thread so we don't block the asyncio loop.""" with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as tmp: tmp.write(audio_bytes) tmp_path = tmp.name try: def _run() -> str: model = _get_whisper_model() # English — reCAPTCHA reads digits in English for most locales. segments, _ = model.transcribe( tmp_path, language="en", beam_size=1, ) text = " ".join(s.text for s in segments).strip() return text.lower() return await asyncio.to_thread(_run) finally: try: Path(tmp_path).unlink(missing_ok=True) except OSError: pass def _normalise_digits(s: str) -> str: """reCAPTCHA expects the digits with no spaces. Whisper often emits `'2 6 1 5'` or word-digits (`'two six one five'`). Normalise both.""" word_to_digit = { "zero": "0", "oh": "0", "one": "1", "two": "2", "three": "3", "four": "4", "five": "5", "six": "6", "seven": "7", "eight": "8", "nine": "9", } out: list[str] = [] tokens = s.replace(",", " ").replace(".", " ").split() for tok in tokens: tok = tok.strip() if not tok: continue if tok.isdigit(): out.append(tok) elif tok in word_to_digit: out.append(word_to_digit[tok]) return "".join(out) async def solve_audio_recaptcha(page, max_attempts: int = 3) -> bool: """Solve whatever reCAPTCHA v2 is embedded on `page`. Returns True if a valid token was issued (visible as non-empty g-recaptcha-response). Handles: - initial checkbox click (sometimes passes silently) - switch to audio mode - Whisper transcription - retry on wrong-answer with "play again" """ # Find the anchor iframe (the "I'm not a robot" checkbox). anchor = page.frame_locator( "iframe[src*='recaptcha/api2/anchor']" ).first try: await anchor.locator("#recaptcha-anchor").click(timeout=8_000) except Exception as e: logger.warning("recaptcha anchor click failed: %r", e) return False # Short settle — sometimes the silent pass happens here. await page.wait_for_timeout(1500) if await _has_token(page): logger.info("recaptcha: silent pass (no challenge needed)") return True # Drive the challenge frame (bframe). bframe = page.frame_locator( "iframe[src*='recaptcha/api2/bframe']" ).first # Switch to audio mode. try: await bframe.locator("#recaptcha-audio-button").click(timeout=8_000) except Exception: logger.warning("recaptcha: audio button not reachable") return False for attempt in range(max_attempts): # Wait for the audio player source to render. try: await bframe.locator("audio#audio-source").wait_for( state="attached", timeout=8_000, ) except Exception: logger.warning("recaptcha attempt %d: audio source never rendered", attempt) continue audio_url = await bframe.locator("audio#audio-source").get_attribute("src") if not audio_url: logger.warning("recaptcha attempt %d: empty audio src", attempt) continue # Download — reCAPTCHA audio is cookie-gated via the bframe origin, # but plain GET works most of the time since the URL is signed. try: async with httpx.AsyncClient(timeout=15.0) as client: r = await client.get(audio_url) r.raise_for_status() audio_bytes = r.content except Exception as e: logger.warning("recaptcha attempt %d: download failed: %r", attempt, e) continue transcript = await _transcribe_audio(audio_bytes) digits = _normalise_digits(transcript) logger.info( "recaptcha attempt %d: transcript=%r → digits=%r", attempt, transcript, digits, ) if not digits: continue try: await bframe.locator("#audio-response").fill(digits, timeout=5_000) await bframe.locator("#recaptcha-verify-button").click(timeout=5_000) except Exception as e: logger.warning("recaptcha attempt %d: submit failed: %r", attempt, e) continue await page.wait_for_timeout(2000) if await _has_token(page): logger.info("recaptcha: solved on attempt %d", attempt + 1) return True # Check if the site flagged wrong answer (error message inside bframe). # Clicking the reload button fetches a new clip. try: await bframe.locator("#recaptcha-reload-button").click(timeout=3_000) await page.wait_for_timeout(1500) except Exception: pass logger.warning("recaptcha: all %d attempts failed", max_attempts) return False async def _has_token(page) -> bool: """Check if the page's g-recaptcha-response textarea has a non-empty value (= challenge was passed).""" val = await page.evaluate( """() => { const el = document.querySelector('textarea[name="g-recaptcha-response"]'); return el ? (el.value || '').length > 0 : false; }""" ) return bool(val)