Luciabrightcode
32179b2190
- Implement GhostCursorEngine with Bezier curves/Fitts's Law in src/browser/ghost_cursor.py - Implement EntropyScheduler (Gaussian jitter/phase drift) in src/core/scheduler.py - Implement MobileProxyRotator (sticky sessions) in src/core/proxy.py - Update CamoufoxManager to target Chrome 124 for TLS consistency - Add manual TLS verification script (tests/manual/verify_tls.py) - Update implementation plan and walkthrough documentation
111 lines
4.1 KiB
Python
111 lines
4.1 KiB
Python
import asyncio
|
|
import json
|
|
import logging
|
|
from src.browser.manager import CamoufoxManager
|
|
from src.extractor.client import CurlClient
|
|
from src.core.session import SessionState
|
|
|
|
# Configure logging
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger("TLSVerifier")
|
|
|
|
TARGET_URL = "https://tls.peet.ws/api/all"
|
|
|
|
async def verify_tls():
|
|
logger.info("Starting TLS Verification Protocol...")
|
|
|
|
# 1. Browser Baseline
|
|
logger.info("Step 1: Capturing Browser Baseline...")
|
|
browser_ja3 = None
|
|
browser_ua = None
|
|
|
|
try:
|
|
async with CamoufoxManager() as browser:
|
|
# Navigate to TLS inspection
|
|
# Note: We need to extract the JSON body from the page
|
|
# Camoufox/Playwright: page.content() or evaluate
|
|
page = await browser.context.new_page()
|
|
await page.goto(TARGET_URL)
|
|
content = await page.evaluate("() => document.body.innerText")
|
|
|
|
try:
|
|
# Debug content if needed
|
|
# logger.info(f"Page Content: {content[:100]}...")
|
|
|
|
data = json.loads(content)
|
|
browser_ja3 = data.get('ja3_hash', 'UNKNOWN')
|
|
browser_ua = data.get('user_agent', 'UNKNOWN')
|
|
logger.info(f"Browser JA3: {browser_ja3}")
|
|
logger.info(f"Browser UA: {browser_ua}")
|
|
|
|
if browser_ja3 == 'UNKNOWN':
|
|
logger.warning(f"Full Content: {content}")
|
|
|
|
# Extraction might fail on some pages (Access Denied for localStorage)
|
|
# We catch it here to continue the TLS test
|
|
try:
|
|
session_state = await browser.extract_session_state()
|
|
logger.info("Session extracted successfully")
|
|
except Exception as e:
|
|
logger.warning(f"Session extraction failed ({e}), using synthetic session for Client phase")
|
|
# Construct synthetic session
|
|
from typing import List, Dict
|
|
session_state = SessionState(
|
|
cookies=[],
|
|
local_storage={},
|
|
session_storage={},
|
|
cf_clearance=None,
|
|
user_agent=browser_ua if browser_ua != 'UNKNOWN' else "Mozilla/5.0 ...",
|
|
tls_fingerprint="chrome120", # Default baseline to test
|
|
timestamp=0
|
|
)
|
|
|
|
except json.JSONDecodeError:
|
|
logger.error("Failed to parse Browser response as JSON")
|
|
logger.debug(content)
|
|
return
|
|
|
|
except Exception as e:
|
|
logger.error(f"Browser Phase Failed: {e}")
|
|
return
|
|
|
|
# 2. Extractor Comparison
|
|
logger.info("Step 2: Capturing Extractor Fingerprint...")
|
|
client_ja3 = None
|
|
client_ua = None
|
|
|
|
try:
|
|
# Use the session state from browser to ensure same UA/headers context
|
|
async with CurlClient(session_state) as client:
|
|
response = await client.fetch(TARGET_URL)
|
|
data = response.json()
|
|
|
|
client_ja3 = data.get('ja3_hash', 'UNKNOWN')
|
|
client_ua = data.get('user_agent', 'UNKNOWN')
|
|
|
|
logger.info(f"Client JA3: {client_ja3}")
|
|
logger.info(f"Client UA: {client_ua}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Extractor Phase Failed: {e}")
|
|
return
|
|
|
|
# 3. Verification
|
|
logger.info("-" * 40)
|
|
logger.info("VERIFICATION RESULTS")
|
|
logger.info("-" * 40)
|
|
|
|
match_ja3 = (browser_ja3 == client_ja3)
|
|
match_ua = (browser_ua == client_ua)
|
|
|
|
logger.info(f"JA3 Match: {'PASS' if match_ja3 else 'FAIL'}")
|
|
logger.info(f"UA Match: {'PASS' if match_ua else 'FAIL'}")
|
|
|
|
if not match_ja3:
|
|
logger.warning(f"Mismatch Detected! Browser: {browser_ja3} != Client: {client_ja3}")
|
|
|
|
if not match_ua:
|
|
logger.warning(f"Mismatch Detected! Browser: {browser_ua} != Client: {client_ua}")
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(verify_tls())
|