""" Authentication tools: form login, browser login, OAuth """ import asyncio import json import re import time from typing import Any, Dict, List from urllib.parse import urljoin from bs4 import BeautifulSoup from mcp.types import TextContent from ..config import ( PLAYWRIGHT_AVAILABLE, authenticated_session, browser_instance, logger, playwright_instance, session, ) from ..session.manager import update_authenticated_session from ..utils.detection import detect_login_error_selectors, detect_protection_indicators async def login_form_tool(args: dict[str, Any]) -> list[TextContent]: """Login through web form with comprehensive error analysis and protection detection.""" login_url = args["login_url"] username = args["username"] password = args["password"] username_field = args.get("username_field") password_field = args.get("password_field") additional_fields = args.get("additional_fields", {}) submit_url = args.get("submit_url") verbose = args.get("verbose", True) retry_strategies = args.get("retry_strategies", []) debug = args.get("debug", False) try: redirect_chain = [] form_analysis = {} protection_detected = None error_details = None login_errors = [] # First, get the login page to find the form logger.info(f"Fetching login page: {login_url}") from ..utils.http_client import get_http_client http_client = await get_http_client() result_data = await http_client.get(login_url, timeout=10) if result_data.get("status_code", 200) >= 400: return [ TextContent( type="text", text=f"HTTP {result_data.get('status_code')}: Failed to fetch {login_url}", ) ] redirect_chain.append( { "url": login_url, "status": result_data.get("status_code", 200), "final_url": result_data.get("final_url", login_url), } ) soup = BeautifulSoup(result_data.get("html", ""), "lxml") # Detect protection mechanisms on login page protection_indicators = detect_protection_indicators() for protection_type, selectors in protection_indicators.items(): for selector in selectors: if soup.select(selector): protection_detected = protection_type error_details = f"{protection_type} detected on login page" break if protection_detected: break # Find login form (look for forms with password inputs) login_forms = [] for form in soup.find_all("form"): if form.find("input", {"type": "password"}): form_info = { "action": form.get("action", ""), "method": form.get("method", "GET").upper(), "inputs": [], } for inp in form.find_all("input"): input_info = { "name": inp.get("name"), "type": inp.get("type", "text"), "id": inp.get("id"), "required": inp.has_attr("required"), "value": inp.get("value", "") if inp.get("type") != "password" else "***", } form_info["inputs"].append(input_info) login_forms.append(form_info) if not login_forms: return [ TextContent( type="text", text=json.dumps( { "error": "No login form found on the page", "login_url": login_url, "final_url": result_data.get("final_url", login_url), "protection_detected": protection_detected, "page_content_snippet": result_data.get("html", "")[:500] + "..." if len(result_data.get("html", "")) > 500 else result_data.get("html", ""), "diagnostics": { "forms_found": len(soup.find_all("form")), "password_inputs_found": len( soup.find_all("input", {"type": "password"}) ), "redirect_chain": redirect_chain, }, }, indent=2, ensure_ascii=False, ), ) ] # Use the first form with password field login_form = soup.find("form").find_parent().find("form") for form in soup.find_all("form"): if form.find("input", {"type": "password"}): login_form = form break form_analysis = login_forms[0] # Store analysis of first form # Auto-detect field names if not provided if not username_field: # Common username field names username_input = ( login_form.find("input", {"type": "email"}) or login_form.find("input", {"name": re.compile(r"(user|email|login)", re.I)}) or login_form.find("input", {"id": re.compile(r"(user|email|login)", re.I)}) or login_form.find("input", {"type": "text"}) # Fallback to first text input ) username_field = username_input.get("name") if username_input else "username" if not password_field: password_input = login_form.find("input", {"type": "password"}) password_field = password_input.get("name") if password_input else "password" # Build form data form_data = {username_field: username, password_field: password} # Add hidden fields (including CSRF tokens) hidden_fields = {} for input_field in login_form.find_all("input", {"type": "hidden"}): name = input_field.get("name") value = input_field.get("value", "") if name: form_data[name] = value hidden_fields[name] = value # Add additional fields form_data.update(additional_fields) # Determine submit URL if not submit_url: form_action = login_form.get("action", "") submit_url = urljoin(login_url, form_action) if form_action else login_url # Submit login form logger.info(f"Submitting login form to: {submit_url}") # Get initial cookies count from browser session initial_cookies_count = 0 if authenticated_session.get("active"): initial_cookies_count = len(authenticated_session.get("browser_cookies", [])) # Store initial response time start_time = time.time() login_result = await http_client.post(submit_url, data=form_data, timeout=10) response_time = time.time() - start_time if login_result.get("status_code", 200) >= 400: return [ TextContent( type="text", text=f"HTTP {login_result.get('status_code')}: Login failed for {submit_url}", ) ] # Track redirect chain for login submission login_redirect_chain = [ { "url": submit_url, "status": login_result.get("status_code", 200), "final_url": login_result.get("final_url", submit_url), } ] # Analyze response content for error indicators response_soup = BeautifulSoup(login_result.get("html", ""), "lxml") # Look for error messages error_selectors = detect_login_error_selectors() for selector in error_selectors: for element in response_soup.select(selector): text = element.get_text(strip=True) if text and len(text) < 300: # Skip very long texts login_errors.append(text) # Detect protection on response page response_protection = None for protection_type, selectors in protection_indicators.items(): for selector in selectors: if response_soup.select(selector): response_protection = protection_type break if response_protection: break # Get current cookies count from browser session current_cookies_count = initial_cookies_count if authenticated_session.get("active"): current_cookies_count = len(authenticated_session.get("browser_cookies", [])) # Analyze response for success/failure indicators success_indicators = { "status_200": login_result.get("status_code", 200) == 200, "status_redirect": 300 <= login_result.get("status_code", 200) < 400, "url_changed": login_result.get("final_url", submit_url) != submit_url, "dashboard_url": "dashboard" in login_result.get("final_url", submit_url).lower(), "profile_url": "profile" in login_result.get("final_url", submit_url).lower(), "home_url": "home" in login_result.get("final_url", submit_url).lower() and "login" not in login_result.get("final_url", submit_url).lower(), "logout_text": "logout" in login_result.get("html", "").lower(), "signout_text": "sign out" in login_result.get("html", "").lower(), "welcome_text": "welcome" in login_result.get("html", "").lower(), "cookies_received": current_cookies_count > initial_cookies_count, "no_login_form": not response_soup.find( "form", lambda x: x and x.find("input", {"type": "password"}) ), "no_errors": len(login_errors) == 0, } failure_indicators = { "error_messages": len(login_errors) > 0, "login_form_present": bool( response_soup.find("form", lambda x: x and x.find("input", {"type": "password"})) ), "unauthorized_status": login_result.get("status_code", 200) == 401, "forbidden_status": login_result.get("status_code", 200) == 403, "login_url_returned": "login" in login_result.get("final_url", submit_url).lower(), "captcha_detected": response_protection == "CAPTCHA", "rate_limited": response_protection == "Rate Limiting", } # Calculate success probability success_score = sum(success_indicators.values()) failure_score = sum(failure_indicators.values()) likely_success = success_score > failure_score and success_score >= 3 # Determine error type based on response if login_result.get("status_code", 200) == 403: if response_protection == "CAPTCHA": error_details = "403 Forbidden - CAPTCHA verification required" elif response_protection == "Rate Limiting": error_details = "403 Forbidden - Rate limiting detected" elif response_protection == "Bot Detection": error_details = "403 Forbidden - Bot detection active" else: error_details = "403 Forbidden - Possible bot protection or invalid credentials" elif login_result.get("status_code", 200) == 401: error_details = "401 Unauthorized - Invalid credentials or authentication required" elif login_errors: error_details = f"Form validation errors: {'; '.join(login_errors[:2])}" elif failure_indicators["login_form_present"] and not likely_success: error_details = "Login form still present - credentials may be invalid" cookies_info = {} for cookie in session.cookies: cookies_info[cookie.name] = { "value": cookie.value[:50] + "..." if len(cookie.value) > 50 else cookie.value, "domain": cookie.domain, "path": cookie.path, "secure": cookie.secure, } # Update authenticated session if login was successful if likely_success: update_authenticated_session( login_url=login_url, cookies=[ { "name": cookie.name, "value": cookie.value, "domain": cookie.domain, "path": cookie.path, "secure": cookie.secure, } for cookie in session.cookies ], session_data={ "login_method": "form", "username": username, "form_fields": list(form_data.keys()), "final_url": login_result.get("final_url", submit_url), "response_time": response_time, }, ) # Build comprehensive result result = { "login_url": login_url, "submit_url": submit_url, "final_url": login_result.get("final_url", submit_url), "status_code": login_result.get("status_code", 200), "response_time_seconds": round(response_time, 2), "likely_success": likely_success, "success_score": f"{success_score}/{len(success_indicators)}", "failure_score": f"{failure_score}/{len(failure_indicators)}", "cookies_received": len(cookies_info), "cookies_before": initial_cookies_count, "cookies_after": len(session.cookies), "form_data_sent": { k: "***" if k == password_field else v for k, v in form_data.items() }, "session_updated": likely_success, "diagnostics": { "error_details": error_details, "protection_detected": protection_detected, "response_protection": response_protection, "login_errors": login_errors, "redirect_chain": redirect_chain + login_redirect_chain, "form_analysis": form_analysis, "hidden_fields": hidden_fields, "success_indicators": {k: v for k, v in success_indicators.items() if v} if verbose else {}, "failure_indicators": {k: v for k, v in failure_indicators.items() if v} if verbose else {}, "response_headers": login_result.get("response_headers", {}) if verbose else {}, "response_content_snippet": login_result.get("html", "")[:1000] + "..." if verbose and len(login_result.get("html", "")) > 1000 else login_result.get("html", "")[:1000] if verbose else None, "cookies_details": cookies_info if verbose else {}, }, "recommendations": [], } # Add specific recommendations based on detected issues if login_result.get("status_code", 200) == 403: result["recommendations"].extend( [ "403 Forbidden suggests bot protection. Try browser_login with manual interaction.", "Import real browser cookies before login attempt.", "Use different User-Agent or anti-detection measures.", ] ) elif response_protection == "CAPTCHA": result["recommendations"].extend( [ "CAPTCHA detected. Manual solving required or anti-captcha service.", "Use browser_login with headless=false for manual CAPTCHA solving.", ] ) elif response_protection == "Rate Limiting": result["recommendations"].extend( [ "Rate limiting detected. Wait before retry or use different IP.", "Implement exponential backoff retry strategy.", ] ) elif login_errors: result["recommendations"].extend( [ "Form validation errors detected. Check username/password format.", "Verify required fields are included in form submission.", "Check if additional verification (email, phone) is required.", ] ) elif not likely_success and len(session.cookies) == initial_cookies_count: result["recommendations"].extend( [ "No new cookies received. Login likely failed.", "Verify credentials are correct.", "Check if JavaScript is required for login process.", ] ) elif failure_indicators["login_form_present"]: result["recommendations"].extend( [ "Login form still present after submission. Check credentials.", "Verify form field names are correct.", "Check if additional form fields (CSRF tokens) are required.", ] ) return [TextContent(type="text", text=json.dumps(result, indent=2, ensure_ascii=False))] except Exception as e: error_result = { "login_url": login_url, "error": str(e), "error_type": type(e).__name__, "form_data_attempted": { k: "***" if "password_field" in locals() and k == password_field else v for k, v in form_data.items() } if "form_data" in locals() else {}, "diagnostics": { "error_details": f"Critical error during form login: {str(e)}", "redirect_chain": redirect_chain if "redirect_chain" in locals() else [], "protection_detected": protection_detected if "protection_detected" in locals() else None, }, "recommendations": [ "Check if login URL is accessible and returns valid HTML", "Verify form field names and requirements", "Try browser_login for complex login flows", "Check if JavaScript is required for form submission", ], } return [ TextContent(type="text", text=json.dumps(error_result, indent=2, ensure_ascii=False)) ] # The browser_login_tool is very large, so I'll continue in a separate write operation...