| import logging
|
| import platform
|
| import sys
|
| import time
|
| from datetime import timedelta
|
| from html import escape
|
| from urllib.parse import unquote, quote
|
|
|
| from func_timeout import FunctionTimedOut, func_timeout
|
| from selenium.common import TimeoutException
|
| from selenium.webdriver.chrome.webdriver import WebDriver
|
| 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, staleness_of, title_is)
|
| from selenium.webdriver.common.action_chains import ActionChains
|
| from selenium.webdriver.support.wait import WebDriverWait
|
|
|
| import utils
|
| from dtos import (STATUS_ERROR, STATUS_OK, ChallengeResolutionResultT,
|
| ChallengeResolutionT, HealthResponse, IndexResponse,
|
| V1RequestBase, V1ResponseBase)
|
| from sessions import SessionsStorage
|
|
|
| ACCESS_DENIED_TITLES = [
|
|
|
| 'Access denied',
|
|
|
| 'Attention Required! | Cloudflare'
|
| ]
|
| ACCESS_DENIED_SELECTORS = [
|
|
|
| 'div.cf-error-title span.cf-code-label span',
|
|
|
| '#cf-error-details div.cf-error-overview h1'
|
| ]
|
| CHALLENGE_TITLES = [
|
|
|
| 'Just a moment...',
|
|
|
| 'DDoS-Guard'
|
| ]
|
| CHALLENGE_SELECTORS = [
|
|
|
| '#cf-challenge-running', '.ray_id', '.attack-box', '#cf-please-wait', '#challenge-spinner', '#trk_jschal_js', '#turnstile-wrapper', '.lds-ring',
|
|
|
| 'td.info #js_info',
|
|
|
| 'div.vc div.text-box h2'
|
| ]
|
|
|
| TURNSTILE_SELECTORS = [
|
| "input[name='cf-turnstile-response']"
|
| ]
|
|
|
| SHORT_TIMEOUT = 1
|
| SESSIONS_STORAGE = SessionsStorage()
|
|
|
|
|
| def test_browser_installation():
|
| logging.info("Testing web browser installation...")
|
| logging.info("Platform: " + platform.platform())
|
|
|
| chrome_exe_path = utils.get_chrome_exe_path()
|
| if chrome_exe_path is None:
|
| logging.error("Chrome / Chromium web browser not installed!")
|
| sys.exit(1)
|
| else:
|
| logging.info("Chrome / Chromium path: " + chrome_exe_path)
|
|
|
| chrome_major_version = utils.get_chrome_major_version()
|
| if chrome_major_version == '':
|
| logging.error("Chrome / Chromium version not detected!")
|
| sys.exit(1)
|
| else:
|
| logging.info("Chrome / Chromium major version: " + chrome_major_version)
|
|
|
| logging.info("Launching web browser...")
|
| user_agent = utils.get_user_agent()
|
| logging.info("FlareSolverr User-Agent: " + user_agent)
|
| logging.info("Test successful!")
|
|
|
|
|
| def index_endpoint() -> IndexResponse:
|
| res = IndexResponse({})
|
| res.msg = "FlareSolverr is ready!"
|
| res.version = utils.get_flaresolverr_version()
|
| res.userAgent = utils.get_user_agent()
|
| return res
|
|
|
|
|
| def health_endpoint() -> HealthResponse:
|
| res = HealthResponse({})
|
| res.status = STATUS_OK
|
| return res
|
|
|
|
|
| def controller_v1_endpoint(req: V1RequestBase) -> V1ResponseBase:
|
| start_ts = int(time.time() * 1000)
|
| logging.info(f"Incoming request => POST /v1 body: {utils.object_to_dict(req)}")
|
| res: V1ResponseBase
|
| try:
|
| res = _controller_v1_handler(req)
|
| except Exception as e:
|
| res = V1ResponseBase({})
|
| res.__error_500__ = True
|
| res.status = STATUS_ERROR
|
| res.message = "Error: " + str(e)
|
| logging.error(res.message)
|
|
|
| res.startTimestamp = start_ts
|
| res.endTimestamp = int(time.time() * 1000)
|
| res.version = utils.get_flaresolverr_version()
|
| logging.debug(f"Response => POST /v1 body: {utils.object_to_dict(res)}")
|
| logging.info(f"Response in {(res.endTimestamp - res.startTimestamp) / 1000} s")
|
| return res
|
|
|
|
|
| def _controller_v1_handler(req: V1RequestBase) -> V1ResponseBase:
|
|
|
| if req.cmd is None:
|
| raise Exception("Request parameter 'cmd' is mandatory.")
|
| if req.headers is not None:
|
| logging.warning("Request parameter 'headers' was removed in FlareSolverr v2.")
|
| if req.userAgent is not None:
|
| logging.warning("Request parameter 'userAgent' was removed in FlareSolverr v2.")
|
|
|
|
|
| if req.maxTimeout is None or int(req.maxTimeout) < 1:
|
| req.maxTimeout = 60000
|
|
|
|
|
| res: V1ResponseBase
|
| if req.cmd == 'sessions.create':
|
| res = _cmd_sessions_create(req)
|
| elif req.cmd == 'sessions.list':
|
| res = _cmd_sessions_list(req)
|
| elif req.cmd == 'sessions.destroy':
|
| res = _cmd_sessions_destroy(req)
|
| elif req.cmd == 'request.get':
|
| res = _cmd_request_get(req)
|
| elif req.cmd == 'request.post':
|
| res = _cmd_request_post(req)
|
| else:
|
| raise Exception(f"Request parameter 'cmd' = '{req.cmd}' is invalid.")
|
|
|
| return res
|
|
|
|
|
| def _cmd_request_get(req: V1RequestBase) -> V1ResponseBase:
|
|
|
| if req.url is None:
|
| raise Exception("Request parameter 'url' is mandatory in 'request.get' command.")
|
| if req.postData is not None:
|
| raise Exception("Cannot use 'postBody' when sending a GET request.")
|
| if req.returnRawHtml is not None:
|
| logging.warning("Request parameter 'returnRawHtml' was removed in FlareSolverr v2.")
|
| if req.download is not None:
|
| logging.warning("Request parameter 'download' was removed in FlareSolverr v2.")
|
|
|
| challenge_res = _resolve_challenge(req, 'GET')
|
| res = V1ResponseBase({})
|
| res.status = challenge_res.status
|
| res.message = challenge_res.message
|
| res.solution = challenge_res.result
|
| return res
|
|
|
|
|
| def _cmd_request_post(req: V1RequestBase) -> V1ResponseBase:
|
|
|
| if req.postData is None:
|
| raise Exception("Request parameter 'postData' is mandatory in 'request.post' command.")
|
| if req.returnRawHtml is not None:
|
| logging.warning("Request parameter 'returnRawHtml' was removed in FlareSolverr v2.")
|
| if req.download is not None:
|
| logging.warning("Request parameter 'download' was removed in FlareSolverr v2.")
|
|
|
| challenge_res = _resolve_challenge(req, 'POST')
|
| res = V1ResponseBase({})
|
| res.status = challenge_res.status
|
| res.message = challenge_res.message
|
| res.solution = challenge_res.result
|
| return res
|
|
|
|
|
| def _cmd_sessions_create(req: V1RequestBase) -> V1ResponseBase:
|
| logging.debug("Creating new session...")
|
|
|
| session, fresh = SESSIONS_STORAGE.create(session_id=req.session, proxy=req.proxy)
|
| session_id = session.session_id
|
|
|
| if not fresh:
|
| return V1ResponseBase({
|
| "status": STATUS_OK,
|
| "message": "Session already exists.",
|
| "session": session_id
|
| })
|
|
|
| return V1ResponseBase({
|
| "status": STATUS_OK,
|
| "message": "Session created successfully.",
|
| "session": session_id
|
| })
|
|
|
|
|
| def _cmd_sessions_list(req: V1RequestBase) -> V1ResponseBase:
|
| session_ids = SESSIONS_STORAGE.session_ids()
|
|
|
| return V1ResponseBase({
|
| "status": STATUS_OK,
|
| "message": "",
|
| "sessions": session_ids
|
| })
|
|
|
|
|
| def _cmd_sessions_destroy(req: V1RequestBase) -> V1ResponseBase:
|
| session_id = req.session
|
| existed = SESSIONS_STORAGE.destroy(session_id)
|
|
|
| if not existed:
|
| raise Exception("The session doesn't exist.")
|
|
|
| return V1ResponseBase({
|
| "status": STATUS_OK,
|
| "message": "The session has been removed."
|
| })
|
|
|
|
|
| def _resolve_challenge(req: V1RequestBase, method: str) -> ChallengeResolutionT:
|
| timeout = int(req.maxTimeout) / 1000
|
| driver = None
|
| try:
|
| if req.session:
|
| session_id = req.session
|
| ttl = timedelta(minutes=req.session_ttl_minutes) if req.session_ttl_minutes else None
|
| session, fresh = SESSIONS_STORAGE.get(session_id, ttl)
|
|
|
| if fresh:
|
| logging.debug(f"new session created to perform the request (session_id={session_id})")
|
| else:
|
| logging.debug(f"existing session is used to perform the request (session_id={session_id}, "
|
| f"lifetime={str(session.lifetime())}, ttl={str(ttl)})")
|
|
|
| driver = session.driver
|
| else:
|
| driver = utils.get_webdriver(req.proxy)
|
| logging.debug('New instance of webdriver has been created to perform the request')
|
| return func_timeout(timeout, _evil_logic, (req, driver, method))
|
| except FunctionTimedOut:
|
| raise Exception(f'Error solving the challenge. Timeout after {timeout} seconds.')
|
| except Exception as e:
|
| raise Exception('Error solving the challenge. ' + str(e).replace('\n', '\\n'))
|
| finally:
|
| if not req.session and driver is not None:
|
| if utils.PLATFORM_VERSION == "nt":
|
| driver.close()
|
| driver.quit()
|
| logging.debug('A used instance of webdriver has been destroyed')
|
|
|
|
|
| def click_verify(driver: WebDriver, num_tabs: int = 1):
|
| try:
|
| logging.debug("Try to find the Cloudflare verify checkbox...")
|
| actions = ActionChains(driver)
|
| actions.pause(5)
|
| for _ in range(num_tabs):
|
| actions.send_keys(Keys.TAB).pause(0.1)
|
| actions.pause(1)
|
| actions.send_keys(Keys.SPACE).perform()
|
|
|
| logging.debug(f"Cloudflare verify checkbox clicked after {num_tabs} tabs!")
|
| except Exception:
|
| logging.debug("Cloudflare verify checkbox not found on the page.")
|
| finally:
|
| driver.switch_to.default_content()
|
|
|
| try:
|
| logging.debug("Try to find the Cloudflare 'Verify you are human' button...")
|
| button = driver.find_element(
|
| by=By.XPATH,
|
| value="//input[@type='button' and @value='Verify you are human']",
|
| )
|
| if button:
|
| actions = ActionChains(driver)
|
| actions.move_to_element_with_offset(button, 5, 7)
|
| actions.click(button)
|
| actions.perform()
|
| logging.debug("The Cloudflare 'Verify you are human' button found and clicked!")
|
| except Exception:
|
| logging.debug("The Cloudflare 'Verify you are human' button not found on the page.")
|
|
|
| time.sleep(2)
|
|
|
| def _get_turnstile_token(driver: WebDriver, tabs: int):
|
| token_input = driver.find_element(By.CSS_SELECTOR, "input[name='cf-turnstile-response']")
|
| current_value = token_input.get_attribute("value")
|
| while True:
|
| click_verify(driver, num_tabs=tabs)
|
| turnstile_token = token_input.get_attribute("value")
|
| if turnstile_token:
|
| if turnstile_token != current_value:
|
| logging.info(f"Turnstile token: {turnstile_token}")
|
| return turnstile_token
|
| logging.debug(f"Failed to extract token possibly click failed")
|
|
|
|
|
| driver.execute_script("""
|
| let el = document.createElement('button');
|
| el.style.position='fixed';
|
| el.style.top='0';
|
| el.style.left='0';
|
| document.body.prepend(el);
|
| el.focus();
|
| """)
|
| time.sleep(1)
|
|
|
| def _resolve_turnstile_captcha(req: V1RequestBase, driver: WebDriver):
|
| turnstile_token = None
|
| if req.tabs_till_verify is not None:
|
| logging.debug(f'Navigating to... {req.url} in order to pass the turnstile challenge')
|
| driver.get(req.url)
|
|
|
| turnstile_challenge_found = False
|
| for selector in TURNSTILE_SELECTORS:
|
| found_elements = driver.find_elements(By.CSS_SELECTOR, selector)
|
| if len(found_elements) > 0:
|
| turnstile_challenge_found = True
|
| logging.info("Turnstile challenge detected. Selector found: " + selector)
|
| break
|
| if turnstile_challenge_found:
|
| turnstile_token = _get_turnstile_token(driver=driver, tabs=req.tabs_till_verify)
|
| else:
|
| logging.debug(f'Turnstile challenge not found')
|
| return turnstile_token
|
|
|
| def _evil_logic(req: V1RequestBase, driver: WebDriver, method: str) -> ChallengeResolutionT:
|
| res = ChallengeResolutionT({})
|
| res.status = STATUS_OK
|
| res.message = ""
|
|
|
|
|
| disable_media = utils.get_config_disable_media()
|
| if req.disableMedia is not None:
|
| disable_media = req.disableMedia
|
| if disable_media:
|
| block_urls = [
|
|
|
| "*.png", "*.jpg", "*.jpeg", "*.gif", "*.webp", "*.bmp", "*.svg", "*.ico",
|
| "*.PNG", "*.JPG", "*.JPEG", "*.GIF", "*.WEBP", "*.BMP", "*.SVG", "*.ICO",
|
| "*.tiff", "*.tif", "*.jpe", "*.apng", "*.avif", "*.heic", "*.heif",
|
| "*.TIFF", "*.TIF", "*.JPE", "*.APNG", "*.AVIF", "*.HEIC", "*.HEIF",
|
|
|
| "*.css",
|
| "*.CSS",
|
|
|
| "*.woff", "*.woff2", "*.ttf", "*.otf", "*.eot",
|
| "*.WOFF", "*.WOFF2", "*.TTF", "*.OTF", "*.EOT"
|
| ]
|
| try:
|
| logging.debug("Network.setBlockedURLs: %s", block_urls)
|
| driver.execute_cdp_cmd("Network.enable", {})
|
| driver.execute_cdp_cmd("Network.setBlockedURLs", {"urls": block_urls})
|
| except Exception:
|
|
|
| logging.debug("Network.setBlockedURLs failed or unsupported on this webdriver")
|
|
|
|
|
| logging.debug(f"Navigating to... {req.url}")
|
| turnstile_token = None
|
|
|
| if method == "POST":
|
| _post_request(req, driver)
|
| else:
|
| if req.tabs_till_verify is None:
|
| driver.get(req.url)
|
| else:
|
| turnstile_token = _resolve_turnstile_captcha(req, driver)
|
|
|
|
|
| if req.cookies is not None and len(req.cookies) > 0:
|
| logging.debug(f'Setting cookies...')
|
| for cookie in req.cookies:
|
| driver.delete_cookie(cookie['name'])
|
| driver.add_cookie(cookie)
|
|
|
| if method == 'POST':
|
| _post_request(req, driver)
|
| else:
|
| driver.get(req.url)
|
|
|
|
|
| if utils.get_config_log_html():
|
| logging.debug(f"Response HTML:\n{driver.page_source}")
|
| html_element = driver.find_element(By.TAG_NAME, "html")
|
| page_title = driver.title
|
|
|
|
|
| for title in ACCESS_DENIED_TITLES:
|
| if page_title.startswith(title):
|
| raise Exception('Cloudflare has blocked this request. '
|
| 'Probably your IP is banned for this site, check in your web browser.')
|
|
|
| for selector in ACCESS_DENIED_SELECTORS:
|
| found_elements = driver.find_elements(By.CSS_SELECTOR, selector)
|
| if len(found_elements) > 0:
|
| raise Exception('Cloudflare has blocked this request. '
|
| 'Probably your IP is banned for this site, check in your web browser.')
|
|
|
|
|
| challenge_found = False
|
| for title in CHALLENGE_TITLES:
|
| if title.lower() == page_title.lower():
|
| challenge_found = True
|
| logging.info("Challenge detected. Title found: " + page_title)
|
| break
|
| if not challenge_found:
|
|
|
| for selector in CHALLENGE_SELECTORS:
|
| found_elements = driver.find_elements(By.CSS_SELECTOR, selector)
|
| if len(found_elements) > 0:
|
| challenge_found = True
|
| logging.info("Challenge detected. Selector found: " + selector)
|
| break
|
|
|
| attempt = 0
|
| if challenge_found:
|
| while True:
|
| try:
|
| attempt = attempt + 1
|
|
|
| for title in CHALLENGE_TITLES:
|
| logging.debug("Waiting for title (attempt " + str(attempt) + "): " + title)
|
| WebDriverWait(driver, SHORT_TIMEOUT).until_not(title_is(title))
|
|
|
|
|
| for selector in CHALLENGE_SELECTORS:
|
| logging.debug("Waiting for selector (attempt " + str(attempt) + "): " + selector)
|
| WebDriverWait(driver, SHORT_TIMEOUT).until_not(
|
| presence_of_element_located((By.CSS_SELECTOR, selector)))
|
|
|
|
|
| break
|
|
|
| except TimeoutException:
|
| logging.debug("Timeout waiting for selector")
|
|
|
| click_verify(driver)
|
|
|
|
|
| html_element = driver.find_element(By.TAG_NAME, "html")
|
|
|
|
|
| logging.debug("Waiting for redirect")
|
|
|
| try:
|
| WebDriverWait(driver, SHORT_TIMEOUT).until(staleness_of(html_element))
|
| except Exception:
|
| logging.debug("Timeout waiting for redirect")
|
|
|
| logging.info("Challenge solved!")
|
| res.message = "Challenge solved!"
|
| else:
|
| logging.info("Challenge not detected!")
|
| res.message = "Challenge not detected!"
|
|
|
| challenge_res = ChallengeResolutionResultT({})
|
| challenge_res.url = driver.current_url
|
| challenge_res.status = 200
|
| challenge_res.cookies = driver.get_cookies()
|
| challenge_res.userAgent = utils.get_user_agent(driver)
|
| challenge_res.turnstile_token = turnstile_token
|
|
|
| if not req.returnOnlyCookies:
|
| challenge_res.headers = {}
|
|
|
| if req.waitInSeconds and req.waitInSeconds > 0:
|
| logging.info("Waiting " + str(req.waitInSeconds) + " seconds before returning the response...")
|
| time.sleep(req.waitInSeconds)
|
|
|
| challenge_res.response = driver.page_source
|
|
|
| if req.returnScreenshot:
|
| challenge_res.screenshot = driver.get_screenshot_as_base64()
|
|
|
| res.result = challenge_res
|
| return res
|
|
|
|
|
| def _post_request(req: V1RequestBase, driver: WebDriver):
|
| post_form = f'<form id="hackForm" action="{req.url}" method="POST">'
|
| query_string = req.postData if req.postData and req.postData[0] != '?' else req.postData[1:] if req.postData else ''
|
| pairs = query_string.split('&')
|
| for pair in pairs:
|
| parts = pair.split('=', 1)
|
|
|
| try:
|
| name = unquote(parts[0])
|
| except Exception:
|
| name = parts[0]
|
| if name == 'submit':
|
| continue
|
|
|
| try:
|
| value = unquote(parts[1]) if len(parts) > 1 else ''
|
| except Exception:
|
| value = parts[1] if len(parts) > 1 else ''
|
|
|
| value=value.replace('"','"')
|
| post_form += f'<input type="text" name="{escape(quote(name))}" value="{escape(quote(value))}"><br>'
|
| post_form += '</form>'
|
| html_content = f"""
|
| <!DOCTYPE html>
|
| <html>
|
| <body>
|
| {post_form}
|
| <script>document.getElementById('hackForm').submit();</script>
|
| </body>
|
| </html>"""
|
| driver.get("data:text/html;charset=utf-8,{html_content}".format(html_content=html_content))
|
|
|
