""" Browser-based authentication using Playwright """ import json import time import asyncio import os from typing import Dict, Any, List from mcp.types import TextContent from ..config import ( PLAYWRIGHT_AVAILABLE, playwright_instance, browser_instance, ensure_browser_session, logger ) from ..utils.detection import detect_protection_indicators, detect_login_error_selectors from ..session.manager import update_authenticated_session async def browser_login_tool(args: Dict[str, Any]) -> List[TextContent]: """Smart browser login with automatic form filling and detailed progress feedback.""" url = args["url"] username = args.get("username") password = args.get("password") headless = args.get("headless", False) wait_for_user = args.get("wait_for_user", True) timeout = args.get("timeout", 300) success_indicator = args.get("success_indicator") username_selector = args.get("username_selector") password_selector = args.get("password_selector") submit_selector = args.get("submit_selector") verbose = args.get("verbose", True) auto_screenshot = args.get("auto_screenshot_on_error", True) # Determine if automatic login should be attempted automatic_login = bool(username and password) if automatic_login: wait_for_user = False # Override wait_for_user if credentials provided if not PLAYWRIGHT_AVAILABLE: return [TextContent(type="text", text=json.dumps({ "error": "Playwright not available. Install with: pip install playwright", "automatic_login": automatic_login, "url": url }, indent=2, ensure_ascii=False))] try: # Progress tracking progress_steps = [] def add_progress(step, status="in_progress", details=None): progress_steps.append({ "step": step, "status": status, "timestamp": time.time(), "details": details }) if verbose: logger.info(f"[BROWSER_LOGIN] {step}: {status}") add_progress("Getting unified browser session", "starting") # Use unified browser session page = await ensure_browser_session() # Collect browser logs browser_logs = [] page.on("console", lambda msg: browser_logs.append(f"[{msg.type}] {msg.text}")) page.on("pageerror", lambda error: browser_logs.append(f"[ERROR] {str(error)}")) page.on("requestfailed", lambda request: browser_logs.append(f"[REQUEST_FAILED] {request.url}")) add_progress("Browser session ready", "completed") # Initialize tracking variables error_details = None element_found = False timeout_reached = False final_screenshot_path = None protection_detected = None login_success = False form_filled = False button_clicked = False # Navigate to login page add_progress("Navigating to login page", "in_progress", url) response = await page.goto(url) await page.wait_for_load_state('domcontentloaded') # Wait for form elements to be ready (critical fix) add_progress("Waiting for form elements to load", "in_progress") await asyncio.sleep(3) # Allow JS to render form elements await page.wait_for_load_state('networkidle', timeout=15000) add_progress("Page loaded", "completed", f"Status: {response.status}") # Check for common protection mechanisms add_progress("Checking for protection mechanisms", "in_progress") protection_indicators = detect_protection_indicators() for protection_type, selectors in protection_indicators.items(): for selector in selectors: try: if await page.locator(selector).count() > 0: protection_detected = protection_type error_details = f"{protection_type} detected on page with selector: {selector}" add_progress("Protection detected", "warning", error_details) break except: continue if protection_detected: break if not protection_detected: add_progress("No protection detected", "completed") # Automatic login flow if automatic_login and not protection_detected: add_progress("Starting automatic login", "in_progress", f"Username: {username[:3]}***") try: # Enhanced auto-detect form fields if selectors not provided if not username_selector: add_progress("Auto-detecting username field", "in_progress") # Comprehensive username field selectors username_candidates = [ "input[type='email']", "input[name*='email' i]", "input[name*='user' i]", "input[name*='login' i]", "input[name='username']", "input[name='userName']", "input[id*='email' i]", "input[id*='user' i]", "input[id*='login' i]", "input[placeholder*='email' i]", "input[placeholder*='username' i]", "input[placeholder*='login' i]", "input[autocomplete='email']", "input[autocomplete='username']", "[data-testid*='email']", "[data-testid*='username']", "[data-test*='email']", "[data-test*='username']", "input[type='text']:first-of-type", "input:not([type='password']):not([type='hidden']):not([type='submit']):not([type='button'])" ] for candidate in username_candidates: try: if await page.locator(candidate).count() > 0: # Verify the field is visible and interactable element = page.locator(candidate).first if await element.is_visible(): username_selector = candidate add_progress("Username field detected", "completed", candidate) break except: continue if not password_selector: add_progress("Auto-detecting password field", "in_progress") # Enhanced password field detection password_candidates = [ "input[type='password']", "input[name*='password' i]", "input[name*='passwd' i]", "input[id*='password' i]", "input[id*='passwd' i]", "input[placeholder*='password' i]", "input[autocomplete='current-password']", "input[autocomplete='password']", "[data-testid*='password']", "[data-test*='password']" ] for candidate in password_candidates: try: if await page.locator(candidate).count() > 0: element = page.locator(candidate).first if await element.is_visible(): password_selector = candidate add_progress("Password field detected", "completed", candidate) break except: continue if not submit_selector: add_progress("Auto-detecting submit button", "in_progress") # Comprehensive submit button selectors submit_candidates = [ "button[type='submit']", "input[type='submit']", "button:has-text('Log in')", "button:has-text('Login')", "button:has-text('Sign in')", "button:has-text('Sign In')", "button:has-text('Enter')", "button:has-text('Submit')", "button:has-text('Войти')", "button:has-text('Вход')", ".login-button", ".submit-button", ".btn-login", ".btn-submit", "[data-testid*='login']", "[data-testid*='submit']", "[data-test*='login']", "[data-test*='submit']", "button:near(input[type='password'])", "form button:last-of-type", "form input[type='submit']" ] for candidate in submit_candidates: try: if await page.locator(candidate).count() > 0: element = page.locator(candidate).first if await element.is_visible() and await element.is_enabled(): submit_selector = candidate add_progress("Submit button detected", "completed", candidate) break except: continue # Final validation of detected selectors detection_result = { "username_selector": username_selector, "password_selector": password_selector, "submit_selector": submit_selector } if not username_selector or not password_selector or not submit_selector: add_progress("Form detection incomplete", "warning", detection_result) # Try to provide diagnostic information try: all_inputs = await page.locator("input").all() all_buttons = await page.locator("button").all() inputs_info = [] for inp in all_inputs[:10]: # Limit to first 10 try: inp_type = await inp.get_attribute("type") or "text" inp_name = await inp.get_attribute("name") or "" inp_id = await inp.get_attribute("id") or "" inp_placeholder = await inp.get_attribute("placeholder") or "" if await inp.is_visible(): inputs_info.append(f"{inp_type}|{inp_name}|{inp_id}|{inp_placeholder}") except: continue buttons_info = [] for btn in all_buttons[:5]: # Limit to first 5 try: btn_type = await btn.get_attribute("type") or "" btn_text = await btn.inner_text() or "" btn_class = await btn.get_attribute("class") or "" if await btn.is_visible() and btn_text.strip(): buttons_info.append(f"{btn_type}|{btn_text.strip()[:20]}|{btn_class}") except: continue add_progress("Available form elements", "info", { "inputs": inputs_info[:5], "buttons": buttons_info[:3] }) except: add_progress("Could not analyze page elements", "warning") else: add_progress("Form fields detected", "completed", detection_result) # Fill username field with proper waiting if username_selector: add_progress("Filling username field", "in_progress", username_selector) try: # Wait for element to be visible and interactable await page.wait_for_selector(username_selector, state="visible", timeout=10000) username_element = page.locator(username_selector).first await username_element.click() # Focus the field await asyncio.sleep(0.5) # Pause before typing await username_element.fill("") # Clear any existing value await username_element.type(username, delay=50) # Type slowly like human add_progress("Username filled", "completed") form_filled = True except Exception as e: add_progress("Username field error", "error", str(e)) error_details = f"Username field error: {str(e)}" form_filled = False else: add_progress("Username field not found", "error") error_details = "Could not locate username field" # Fill password field with proper waiting if password_selector and form_filled: add_progress("Filling password field", "in_progress", password_selector) try: # Wait for element to be visible and interactable await page.wait_for_selector(password_selector, state="visible", timeout=10000) password_element = page.locator(password_selector).first await password_element.click() # Focus the field await asyncio.sleep(0.5) # Pause before typing await password_element.fill("") # Clear any existing value await password_element.type(password, delay=50) # Type slowly like human add_progress("Password filled", "completed") await asyncio.sleep(1) # Pause before submit except Exception as e: add_progress("Password field error", "error", str(e)) error_details = f"Password field error: {str(e)}" form_filled = False else: add_progress("Password field not found", "error") error_details = "Could not locate password field" form_filled = False # Click submit button with proper waiting if submit_selector and form_filled: add_progress("Clicking login button", "in_progress", submit_selector) try: # Wait for button to be clickable await page.wait_for_selector(submit_selector, state="visible", timeout=10000) submit_element = page.locator(submit_selector).first # Ensure button is enabled is_enabled = await submit_element.is_enabled() if not is_enabled: add_progress("Submit button disabled, waiting...", "warning") await asyncio.sleep(2) await submit_element.click() button_clicked = True add_progress("Login button clicked", "completed") # Wait longer for login processing add_progress("Waiting for login response", "in_progress") try: await page.wait_for_load_state('domcontentloaded', timeout=15000) await asyncio.sleep(2) # Additional wait for JS processing add_progress("Page response received", "completed") except: add_progress("No navigation detected, checking current state", "warning") except Exception as e: add_progress("Submit button error", "error", str(e)) error_details = f"Submit button error: {str(e)}" button_clicked = False else: add_progress("Submit button not found", "error") error_details = "Could not locate submit button" except Exception as auto_error: add_progress("Automatic login failed", "error", str(auto_error)) error_details = f"Automatic login error: {str(auto_error)}" form_filled = False button_clicked = False # Manual interaction flow (if needed) elif wait_for_user and not automatic_login and not headless: add_progress("Waiting for manual user interaction", "in_progress", f"Timeout: {timeout}s") start_time = asyncio.get_event_loop().time() while True: current_time = asyncio.get_event_loop().time() elapsed = current_time - start_time if elapsed > timeout: timeout_reached = True error_details = f"Manual login timeout after {timeout} seconds" add_progress("Manual login timeout", "error", error_details) break # Provide periodic progress updates if int(elapsed) % 30 == 0 and elapsed > 0: add_progress("Still waiting for user", "in_progress", f"Elapsed: {int(elapsed)}s / {timeout}s") # Check for success indicator if success_indicator: try: await page.wait_for_selector(success_indicator, timeout=1000) login_success = True element_found = True error_details = None add_progress("Success indicator found", "completed", success_indicator) break except: continue else: # Check if URL changed (common login success indicator) current_url = page.url if current_url != url and "login" not in current_url.lower(): login_success = True add_progress("URL changed - login successful", "completed", current_url) break await asyncio.sleep(1) # Success detection for automatic login if automatic_login and form_filled and button_clicked: add_progress("Detecting login success", "in_progress") # Wait a bit for the login to process await asyncio.sleep(2) current_url = page.url if success_indicator: try: await page.wait_for_selector(success_indicator, timeout=5000) login_success = True element_found = True add_progress("Success indicator found", "completed", success_indicator) except: add_progress("Success indicator not found", "warning") # Additional success checks if not login_success: # Check URL change if current_url != url and "login" not in current_url.lower(): login_success = True add_progress("URL changed - login successful", "completed", current_url) # Check for logout/profile elements logout_indicators = [".logout", ".sign-out", ".profile", ".user-menu", ".dashboard"] for indicator in logout_indicators: if await page.locator(indicator).count() > 0: login_success = True add_progress("Profile elements found - login successful", "completed", indicator) break # Analyze final page state final_url = page.url page_title = await page.title() # Look for login failure indicators add_progress("Analyzing login result", "in_progress") login_failure_indicators = detect_login_error_selectors() login_errors = [] for selector in login_failure_indicators: try: elements = await page.locator(selector).all() for element in elements: text = await element.inner_text() if text.strip() and len(text) < 200: login_errors.append(text.strip()) except: continue if login_errors: add_progress("Login errors detected", "error", login_errors[:3]) else: add_progress("No login errors found", "completed") # Take screenshot if needed if (auto_screenshot and (not login_success or error_details or login_errors)) or verbose: try: os.makedirs("./debug", exist_ok=True) timestamp = int(time.time()) final_screenshot_path = f"./debug/browser_login_{timestamp}.png" await page.screenshot(path=final_screenshot_path, full_page=True) add_progress("Screenshot saved", "completed", final_screenshot_path) except Exception as screenshot_error: add_progress("Screenshot failed", "error", str(screenshot_error)) browser_logs.append(f"[SCREENSHOT_ERROR] {str(screenshot_error)}") # Get cookies cookies = await page.context.cookies() add_progress("Cookies extracted", "completed", f"Count: {len(cookies)}") # Save to global authenticated session for compatibility if login_success or cookies: update_authenticated_session( login_url=url, cookies=cookies, session_data={ "automatic_login": automatic_login, "form_filled": form_filled, "button_clicked": button_clicked, "final_url": page.url, "page_title": await page.title(), "unified_browser_session": True # Mark that we're using unified session } ) add_progress("Session saved to unified browser state", "completed", f"All tools now share {len(cookies)} cookies") # Final success determination if automatic_login and not login_success and not error_details: if login_errors: error_details = f"Login errors detected: {'; '.join(login_errors[:3])}" elif final_url == url: error_details = "No URL change detected - login may have failed" elif not cookies: error_details = "No cookies received - login likely failed" # Final progress update if login_success: add_progress("Login completed successfully", "completed") else: add_progress("Login failed or uncertain", "error", error_details) # Build comprehensive result result = { "login_url": url, "final_url": final_url, "page_title": page_title, "success": login_success, "automatic_login": automatic_login, "form_filled": form_filled, "button_clicked": button_clicked, "headless_mode": headless, "timeout": timeout, "timeout_reached": timeout_reached, "element_found": element_found, "success_indicator": success_indicator, "cookies_imported": len(cookies), "session_updated": True, "progress_steps": progress_steps, "diagnostics": { "error_details": error_details, "protection_detected": protection_detected, "login_errors": login_errors, "browser_logs": browser_logs[-15:] if verbose else [], "final_screenshot": final_screenshot_path, "response_status": response.status if response else None, "response_headers": dict(response.headers) if response and verbose else {}, "url_changed": final_url != url, "form_selectors": { "username_selector": username_selector, "password_selector": password_selector, "submit_selector": submit_selector } if automatic_login else {}, "cookies_details": [ { "name": c["name"], "domain": c.get("domain"), "secure": c.get("secure"), "httpOnly": c.get("httpOnly"), "sameSite": c.get("sameSite") } for c in cookies ] if verbose else [] }, "recommendations": [] } # Add specific recommendations if automatic_login and not form_filled: result["recommendations"].extend([ "Automatic form filling failed. Check selectors or try manual login.", "Use custom username_selector, password_selector, submit_selector parameters.", "Inspect page source to verify form field names and structure." ]) elif protection_detected == "CAPTCHA": result["recommendations"].extend([ "CAPTCHA detected. Switch to manual mode with headless=false.", "Remove username/password to enable manual interaction.", "Consider using anti-captcha services for automation." ]) elif protection_detected == "Cloudflare": result["recommendations"].extend([ "Cloudflare protection detected. Try anti_detection=true in fetch_html.", "Use stealth mode with different browser settings.", "Consider rotating IP addresses if rate limited." ]) elif protection_detected == "Bot Detection": result["recommendations"].extend([ "Bot detection active. Use stealth mode or different user agent.", "Try manual login with headless=false.", "Import real browser cookies before login attempt." ]) elif login_errors: result["recommendations"].extend([ "Form validation errors detected. Check credentials format.", "Verify username/email format and password requirements.", "Check if account verification is required." ]) elif timeout_reached: result["recommendations"].extend([ "Login timeout reached. Increase timeout value.", "Check if success_indicator selector is correct.", "Try different success detection methods." ]) elif automatic_login and not login_success: result["recommendations"].extend([ "Automatic login may have failed. Check credentials.", "Try manual login mode for debugging.", "Verify form selectors are correct." ]) # Don't close context - using unified browser session return [TextContent(type="text", text=json.dumps(result, indent=2, ensure_ascii=False))] except Exception as e: # Emergency cleanup and screenshot emergency_screenshot = None try: if 'page' in locals() and auto_screenshot: os.makedirs("./debug", exist_ok=True) emergency_screenshot = f"./debug/browser_login_crash_{int(time.time())}.png" await page.screenshot(path=emergency_screenshot, full_page=True) except: pass # Emergency cleanup - don't close unified browser session pass error_result = { "login_url": url, "success": False, "automatic_login": automatic_login, "error": str(e), "error_type": type(e).__name__, "emergency_screenshot": emergency_screenshot, "progress_steps": progress_steps if 'progress_steps' in locals() else [], "diagnostics": { "error_details": f"Critical error during browser login: {str(e)}", "browser_logs": browser_logs[-10:] if 'browser_logs' in locals() else [], "protection_detected": protection_detected if 'protection_detected' in locals() else None }, "recommendations": [ "Check if Playwright is properly installed: playwright install", "Verify URL is accessible and returns valid HTML", "Try with headless=false for debugging", "Check browser permissions and system resources", "For automatic login, verify username/password parameters", "For manual login, ensure wait_for_user=true and headless=false" ] } return [TextContent(type="text", text=json.dumps(error_result, indent=2, ensure_ascii=False))]