| import json
|
| import os
|
| import glob
|
| from random import uniform, randint
|
| from time import sleep
|
| from urllib.request import urlretrieve
|
|
|
| from pydub import AudioSegment
|
| from pydub.effects import normalize
|
| from selenium import webdriver
|
| from selenium.webdriver.common.action_chains import ActionChains
|
| import undetected_chromedriver as uc
|
| from selenium.common.exceptions import NoSuchElementException, TimeoutException
|
| from selenium.webdriver.common.by import By
|
| from selenium.webdriver.common.keys import Keys
|
| from selenium.webdriver.support.expected_conditions import \
|
| presence_of_element_located as poel
|
| from selenium.webdriver.support.ui import WebDriverWait
|
| from selenium.webdriver.chrome.options import Options
|
| from selenium.webdriver.chrome.service import Service
|
|
|
| from selenium.webdriver.firefox.options import Options as FirefoxOptions
|
| from selenium.webdriver.firefox.service import Service as GeckoService
|
| from webdriver_manager.chrome import ChromeDriverManager
|
| from webdriver_manager.firefox import GeckoDriverManager
|
| from speech_recognition import AudioFile, Recognizer
|
| from faster_whisper import WhisperModel
|
| from user_agent import generate_user_agent
|
|
|
| from colorize import Colors
|
| from proxier import Proxier
|
|
|
|
|
| class RecaptchaBypass:
|
| def __init__(self, debug=False, use_proxy=False, use_agent=False, headless=True):
|
| self.debug = debug
|
| self.use_proxy = use_proxy
|
| self.use_agent = use_agent
|
| self.headless = headless
|
| self.current_proxy = None
|
| self.proxier = Proxier() if use_proxy else None
|
| self._print_message(Colors.INFO, f'{Colors.blue}Loading Whisper model (tiny.en)...{Colors.reset}')
|
| self.whisper_model = WhisperModel("tiny.en", device="cpu", compute_type="int8")
|
| self.browser = self._get_chromedriver(use_proxy, use_agent, headless)
|
| self.wait = WebDriverWait(self.browser, 30)
|
|
|
| def _get_chromedriver(self, use_proxy, use_agent, headless):
|
| camoufox_cache = os.path.expanduser('~/.cache/camoufox')
|
| camoufox_bin = None
|
| if os.path.isdir(camoufox_cache):
|
| candidates = [
|
| os.path.join(camoufox_cache, 'camoufox-bin'),
|
| os.path.join(camoufox_cache, 'camoufox'),
|
| os.path.join(camoufox_cache, 'browser', 'camoufox')
|
| ]
|
| for c in candidates:
|
| if os.path.isfile(c) and os.access(c, os.X_OK):
|
| camoufox_bin = c
|
| break
|
|
|
| if camoufox_bin:
|
| self._print_message(Colors.INFO, f'{Colors.purple}Camoufox binary detected {Colors.red}>{Colors.reset} {camoufox_bin}')
|
| gecko_path = '/usr/bin/geckodriver'
|
| if not os.path.exists(gecko_path):
|
| gecko_path = GeckoDriverManager().install()
|
|
|
| fox_opts = FirefoxOptions()
|
| os.environ['MOZ_NODEJS'] = '1'
|
| if headless:
|
| fox_opts.add_argument('-headless')
|
|
|
| fox_opts.binary_location = camoufox_bin
|
| fox_opts.set_preference('security.sandbox.content.level', 0)
|
| fox_opts.set_preference('dom.webdriver.enabled', False)
|
| fox_opts.set_preference('useprivacy', True)
|
| fox_opts.set_preference('general.useragent.override', generate_user_agent())
|
| fox_opts.set_preference('intl.accept_languages', 'en-US, en')
|
| fox_opts.set_preference('media.volume_scale', '0.0')
|
|
|
| if use_proxy:
|
| proxy_str = self._setup_proxy()
|
| if '://' in proxy_str:
|
| addr = proxy_str.split('://')[1]
|
| host, port = addr.split(':')
|
| fox_opts.set_preference('network.proxy.type', 1)
|
| fox_opts.set_preference('network.proxy.http', host)
|
| fox_opts.set_preference('network.proxy.http_port', int(port))
|
| fox_opts.set_preference('network.proxy.ssl', host)
|
| fox_opts.set_preference('network.proxy.ssl_port', int(port))
|
|
|
| service = GeckoService(gecko_path)
|
| return webdriver.Firefox(service=service, options=fox_opts)
|
|
|
| chrome_options = Options()
|
| if headless: chrome_options.add_argument('--headless=new')
|
| return uc.Chrome(options=chrome_options)
|
|
|
| def _print_message(self, level, message):
|
| print(f'{level} {message}')
|
|
|
| def _quit_browser(self):
|
| try: self.browser.quit()
|
| except: pass
|
|
|
| def _human_mouse_movement(self):
|
| try:
|
| actions = ActionChains(self.browser)
|
| for _ in range(randint(2, 4)):
|
| actions.move_by_offset(randint(-20, 20), randint(-20, 20)).pause(uniform(0.1, 0.2))
|
| actions.perform()
|
| except: pass
|
|
|
| def _solve_logic(self, i):
|
| if self._is_block():
|
| return {'status': 'blocked'}
|
|
|
|
|
| self._print_message(Colors.INFO, f'{Colors.blue}Getting recaptcha checkbox..{Colors.reset}')
|
| iframe = self.wait.until(poel((By.CSS_SELECTOR, "iframe[src*='recaptcha/api2/anchor']")))
|
| self.browser.switch_to.frame(iframe)
|
| check_box = self.wait.until(poel((By.ID, "recaptcha-anchor")))
|
| self._human_mouse_movement()
|
| ActionChains(self.browser).move_to_element(check_box).pause(uniform(0.5, 1.0)).click().perform()
|
| self.browser.switch_to.default_content()
|
|
|
| sleep(uniform(2, 4))
|
|
|
| if self._is_block():
|
| return {'status': 'blocked'}
|
|
|
|
|
| token = self._get_token_from_page()
|
| if token: return {'status': 'solved', 'token': token}
|
|
|
|
|
| self._print_message(Colors.INFO, f'{Colors.blue}Getting recaptcha challenge..{Colors.reset}')
|
| self.wait.until(lambda d: len(d.find_elements(By.CSS_SELECTOR, "iframe[src*='recaptcha/api2/bframe']")) > 0)
|
| iframe = self.browser.find_element(By.CSS_SELECTOR, "iframe[src*='recaptcha/api2/bframe']")
|
| self.browser.switch_to.frame(iframe)
|
|
|
| if self._is_block():
|
| self.browser.switch_to.default_content()
|
| return {'status': 'blocked'}
|
|
|
| audio_button = self.wait.until(poel((By.ID, 'recaptcha-audio-button')))
|
| ActionChains(self.browser).move_to_element(audio_button).pause(uniform(0.5, 1.0)).click().perform()
|
|
|
|
|
| download_link = self.wait.until(poel((By.CLASS_NAME, 'rc-audiochallenge-tdownload-link')))
|
| audio_url = download_link.get_attribute('href')
|
| self.browser.switch_to.default_content()
|
|
|
|
|
| phrase_text, _ = self._audio_to_text(audio_url)
|
| self._print_message(Colors.INFO, f'{Colors.purple}Phrase Text {Colors.red}> {Colors.green}{phrase_text}{Colors.reset}')
|
|
|
|
|
| self.browser.switch_to.frame(self.browser.find_element(By.CSS_SELECTOR, "iframe[src*='recaptcha/api2/bframe']"))
|
| input_field = self.wait.until(poel((By.ID, 'audio-response')))
|
| for char in phrase_text:
|
| input_field.send_keys(char)
|
| sleep(uniform(0.05, 0.15))
|
| sleep(uniform(0.5, 1))
|
| input_field.send_keys(Keys.ENTER)
|
| self.browser.switch_to.default_content()
|
|
|
| sleep(uniform(3, 5))
|
| token = self._get_token_from_page()
|
| if token:
|
| return {'status': 'solved', 'token': token}
|
| return {'status': 'failed'}
|
|
|
| def get_token(self, url, max_retries=5):
|
| for i in range(max_retries):
|
| self._print_message(Colors.INFO, f'{Colors.blue}Token retrieval attempt {i+1}/{max_retries}..{Colors.reset}')
|
| try:
|
| if i > 0:
|
| self._quit_browser()
|
| self.browser = self._get_chromedriver(self.use_proxy, self.use_agent, self.headless)
|
| self.wait = WebDriverWait(self.browser, 30)
|
|
|
| self.browser.get(url)
|
| sleep(uniform(5, 7))
|
| result = self._solve_logic(i)
|
|
|
| if result['status'] == 'solved':
|
| self._print_message(Colors.SUCCESS, f'Token retrieved! 📍\n{Colors.white}{result["token"]}{Colors.reset}\n')
|
| token = result['token']
|
| self._quit_browser()
|
| return {'success': True, 'token': token}
|
| elif result['status'] == 'blocked':
|
| if self.use_proxy and self.current_proxy: self.proxier.remove_proxy(self.current_proxy)
|
| continue
|
| except Exception as e:
|
| self._print_message(Colors.FAILED, f'Error: {str(e)}')
|
| continue
|
| self._quit_browser()
|
| return {'success': False, 'error': 'Max retries reached'}
|
|
|
| def bypass_recaptcha_v2(self, url, max_retries=5):
|
| for i in range(max_retries):
|
| self._print_message(Colors.INFO, f'{Colors.blue}Bypass attempt {i+1}/{max_retries}..{Colors.reset}')
|
| try:
|
| if i > 0:
|
| self._quit_browser()
|
| self.browser = self._get_chromedriver(self.use_proxy, self.use_agent, self.headless)
|
| self.wait = WebDriverWait(self.browser, 30)
|
|
|
| self.browser.get(url)
|
| sleep(uniform(5, 7))
|
| solve_res = self._solve_logic(i)
|
|
|
| if solve_res['status'] == 'solved':
|
| token = solve_res['token']
|
| self._print_message(Colors.SUCCESS, 'V2 Bypassed! 📍')
|
| self._print_message(Colors.INFO, f'Token: {token[:100]}...')
|
|
|
|
|
| if 'google.com/recaptcha/api2/demo' in self.browser.current_url:
|
| try:
|
| submit_btn = self.browser.find_element(By.ID, 'recaptcha-demo-submit')
|
| self.browser.execute_script("arguments[0].click();", submit_btn)
|
| sleep(3)
|
| except: pass
|
|
|
| self._quit_browser()
|
| return {'success': True, 'token': token}
|
|
|
| if solve_res['status'] == 'blocked' and self.use_proxy:
|
| self.proxier.remove_proxy(self.current_proxy)
|
| self._quit_browser()
|
| except Exception as e:
|
| self._print_message(Colors.FAILED, f'Error: {str(e)}')
|
| self._quit_browser()
|
| return {'success': False}
|
|
|
| def _get_token_from_page(self):
|
| try:
|
| textarea = self.browser.find_element(By.ID, "g-recaptcha-response")
|
| token = textarea.get_attribute("value")
|
| if token and len(token) > 100: return token
|
| except: pass
|
| return None
|
|
|
| def _audio_to_text(self, audio_url):
|
| filename = 'audio.mp3'
|
| wav_output = 'audio.wav'
|
| import urllib.request
|
| opener = urllib.request.build_opener()
|
| opener.addheaders = [('User-Agent', generate_user_agent())]
|
| urllib.request.install_opener(opener)
|
| urlretrieve(audio_url, filename)
|
| AudioSegment.from_mp3(filename).export(wav_output, format='wav')
|
| segments, _ = self.whisper_model.transcribe(wav_output)
|
| phrase = " ".join([s.text for s in segments]).strip()
|
| if not self.debug:
|
| for f in [filename, wav_output]:
|
| if os.path.exists(f): os.remove(f)
|
| return phrase, wav_output
|
|
|
| def _is_block(self):
|
| try:
|
| txt = self.browser.page_source.lower()
|
| if "try again later" in txt or "automated queries" in txt: return True
|
| except: pass
|
| return None
|
|
|
