#!/usr/bin/env python3 """ Vibe Pentest — 交互式浏览器登录凭证捕获 借鉴 okscope auth_browser_login.py 的核心逻辑: - 纯倒计时自动提取 (无需 Ctrl+C) - DOM/URL/网络 三重登录检测 - 自动登录页发现 - JWT 自动解码 - 会话固定检测 - 3次失败回退 - CDP 模式 (连接已有浏览器) 用法: python extract_credentials.py --url https://target.com/login python extract_credentials.py --url https://target.com --timeout 120 --username admin --output sessions/admin.json python extract_credentials.py --cdp-port 9222 --output sessions/admin.json """ import argparse import base64 import json import re import sys import time from datetime import datetime, timezone from pathlib import Path from urllib.parse import urlparse # Windows GBK 终端中文乱码修复 if sys.stdout.encoding and sys.stdout.encoding.lower() in ('gbk', 'cp936', 'cp950'): try: sys.stdout.reconfigure(encoding='utf-8', errors='replace') except Exception: pass try: from playwright.sync_api import sync_playwright except ImportError: print("Error: playwright not installed. Run: pip install playwright && playwright install chromium") sys.exit(1) # ── JWT 解码 ────────────────────────────────────────────── def decode_jwt(token: str) -> dict | None: """尝试解码 JWT (不验证签名)，提取 payload""" try: parts = token.split(".") if len(parts) != 3: return None payload_b64 = parts[1] payload_b64 += "=" * (4 - len(payload_b64) % 4) payload_bytes = base64.urlsafe_b64decode(payload_b64) return json.loads(payload_bytes) except Exception: return None def looks_like_jwt(value: str) -> bool: return bool(re.match(r'^[A-Za-z0-9\-_]+\.[A-Za-z0-9\-_]+\.[A-Za-z0-9\-_]+$', value)) # ── DOM 登录检测 ────────────────────────────────────────── def get_login_signals(page, initial_url: str, dom_before: str = None) -> list: """多方面检测登录是否完成，返回信号列表""" signals = [] try: current_url = page.url except Exception: return signals # 1. URL 变化 (跳离登录页) if current_url != initial_url: login_kw = ["/login", "/signin", "/auth", "/sign-in"] if not any(kw in current_url.lower() for kw in login_kw): signals.append(("URL跳转", current_url)) # 2. 页面标题变化 try: title = page.title() if title and "login" not in title.lower() and "登录" not in title: signals.append(("页面标题", title[:80])) except Exception: pass # 3. DOM 元素检测 try: dom_signals_list = page.evaluate("""() => { let result = []; const indicators = [ ['.user-info', '用户信息'], ['.user-avatar', '头像'], ['.avatar', '头像'], ['.username', '用户名'], ['.logout', '退出按钮'], ['#logout', '退出按钮'], ['a[href*="logout"]', '退出链接'], ['.sidebar', '侧边栏'], ['.navbar-user', '导航用户区'], ['.dashboard', '仪表盘'], ['.main-content', '主内容区'], ['.welcome', '欢迎信息'], ['.profile', '用户资料'], ['.admin-panel', '管理面板'], ['.menu', '菜单'], ]; for (let [sel, label] of indicators) { try { let el = document.querySelector(sel); if (el && el.offsetParent !== null) { let text = (el.textContent || '').trim().slice(0, 60); result.push('DOM:' + label + (text ? '=' + text : '')); } } catch(e) {} } // 登录表单是否消失 let loginForm = document.querySelector( 'input[type="password"], form[action*="login"]' ); if (!loginForm || loginForm.offsetParent === null) { result.push('DOM:登录表单已消失'); } return result; }""") signals.extend([("DOM", s) for s in dom_signals_list]) except Exception: pass # 3b. DOM 内容变化对比 (针对 SPA 登录不跳转 URL 的场景) if dom_before: try: dom_after = page.evaluate("""() => { const body = document.body || document.documentElement; const text = (body.innerText || '').trim(); return text.slice(0, 2000); }""") if dom_before != dom_after and len(dom_after) > 0: # 检查页面文本是否出现登录后的特征 post_login_keywords = ['欢迎', 'welcome', '退出', 'logout', '用户中心', 'dashboard', 'profile', '我的', '设置', '消息', '通知', 'admin', '管理', '首页'] has_post_signal = any(kw in dom_after.lower() for kw in [k.lower() for k in post_login_keywords]) has_login_form = any(kw in dom_after.lower() for kw in ['password', '密码', 'login', '登录']) if has_post_signal and not has_login_form: preview = dom_after[:80].replace('\n', ' ') signals.append(("DOM内容变化", preview)) elif has_post_signal: signals.append(("DOM含登录特征", "页面包含登录后内容特征")) except Exception: pass # 4. Token 存在检测 try: has_token = page.evaluate("""() => { if (/token|session|auth|jwt/i.test(document.cookie)) return 'cookie'; try { for (let i = 0; i < localStorage.length; i++) { let k = localStorage.key(i); if (/token|auth|jwt|session/i.test(k) && localStorage.getItem(k)) return 'localStorage:' + k; } } catch(e) {} return null; }""") if has_token: signals.append(("Token检测", has_token)) except Exception: pass return signals # ── 核心凭证提取 ─────────────────────────────────────────── def extract_credentials_from_page(page, cookies_before=None, verify_url=None, screenshot_path=None, role_label=None) -> dict: """从浏览器页面提取所有认证数据""" credentials = { "role_label": role_label or "", "cookies": [], "cookies_before": cookies_before or [], "tokens": [], "storage": {"localStorage": {}, "sessionStorage": {}}, "jwt_analysis": [], "auth_cookie_string": "", "auth_headers": [], "session_fixation": None, "login_url": page.url, "capture_time": datetime.now(timezone.utc).isoformat(), "warnings": [], } # 1. Cookie try: cookies = page.context.cookies() for c in cookies: credentials["cookies"].append({ "name": c["name"], "value": c["value"], "domain": c.get("domain", ""), "path": c.get("path", "/"), "secure": c.get("secure", False), "httpOnly": c.get("httpOnly", False), "sameSite": c.get("sameSite", "None"), }) except Exception as e: print(f" [WARN] Cookie 提取失败: {e}", file=sys.stderr) # 2. Storage try: storage_data = page.evaluate("""() => { const result = { localStorage: {}, sessionStorage: {} }; try { for (let i = 0; i < localStorage.length; i++) { const k = localStorage.key(i); result.localStorage[k] = localStorage.getItem(k); } } catch(e) { result.localStorageError = e.message; } try { for (let i = 0; i < sessionStorage.length; i++) { const k = sessionStorage.key(i); result.sessionStorage[k] = sessionStorage.getItem(k); } } catch(e) { result.sessionStorageError = e.message; } return result; }""") credentials["storage"] = storage_data except Exception as e: print(f" [WARN] Storage 提取失败: {e}", file=sys.stderr) # 3. Token 识别 token_keywords = [ "token", "access_token", "refresh_token", "id_token", "jwt", "authorization", "bearer", "api_key", "apikey", "x-auth-token", "session", "session_id", "sid", "auth", "credential", ] for stype in ["localStorage", "sessionStorage"]: data = credentials.get("storage", {}).get(stype, {}) if isinstance(data, dict): for key, value in data.items(): if any(p in key.lower() for p in token_keywords) and value: credentials["tokens"].append({"type": stype, "key": key, "value": value}) for c in credentials["cookies"]: if any(p in c["name"].lower() for p in token_keywords): credentials["tokens"].append({"type": "cookie", "key": c["name"], "value": c["value"]}) # 4. Cookie 字符串 auth_parts = [f'{c["name"]}={c["value"]}' for c in credentials["cookies"]] credentials["auth_cookie_string"] = "; ".join(auth_parts) # 5. 认证头 auth_headers = [] bearer_token = None for t in credentials["tokens"]: if looks_like_jwt(t["value"]): bearer_token = t["value"] break if not bearer_token: for t in credentials["tokens"]: if t["key"].lower() in ("token", "access_token", "bearer", "authorization"): bearer_token = t["value"] break if bearer_token: auth_headers.append({"type": "header", "name": "Authorization", "value": f"Bearer {bearer_token}"}) if credentials["auth_cookie_string"]: auth_headers.append({"type": "header", "name": "Cookie", "value": credentials["auth_cookie_string"]}) credentials["auth_headers"] = auth_headers # 空值告警 if not credentials["auth_cookie_string"] and not auth_headers: credentials["warnings"].append("未捕获到任何 Cookie 或认证 Token，可能登录未成功") elif not credentials["auth_cookie_string"] and bearer_token: credentials["warnings"].append( "AUTH_COOKIE 为空但发现了 Bearer Token — 扫描工具需使用 Authorization 头而非 Cookie 头" ) # 登录前后 Cookie 完全一致 — 已有 Cookie 激活场景 elif cookies_before: before_map = {c["name"]: c["value"] for c in cookies_before} after_map = {c["name"]: c["value"] for c in credentials["cookies"]} if before_map and after_map and before_map == after_map: credentials["warnings"].append( "登录前后 Cookie 值完全一致，未产生新凭证 — " "可能通过已有 Cookie 状态激活登录，或使用了 HTTP-only Session" ) # 6. JWT 解码 seen_tokens = set() for t in credentials["tokens"]: val = t["value"] if val in seen_tokens: continue seen_tokens.add(val) if looks_like_jwt(val): decoded = decode_jwt(val) if decoded: exp_ts = decoded.get("exp") exp_str = None if exp_ts: try: exp_str = datetime.fromtimestamp(int(exp_ts), tz=timezone.utc).isoformat() except Exception: exp_str = str(exp_ts) analysis = {"source_key": t["key"], "source_type": t["type"], "payload": decoded} if exp_str: analysis["expires_at"] = exp_str try: expires_in = int(exp_ts) - int(time.time()) analysis["expires_in_seconds"] = expires_in if expires_in < 3600: analysis["warning"] = f"Token {expires_in // 60} 分钟后过期" except Exception: pass credentials["jwt_analysis"].append(analysis) # 7. 会话固定检测 if cookies_before: for cb in cookies_before: for ca in credentials["cookies"]: if cb["name"] == ca["name"] and cb["value"] == ca["value"]: credentials["session_fixation"] = { "detected": True, "cookie_name": cb["name"], "detail": f"登录前后 {cb['name']} 未重新生成，存在会话固定风险", } credentials["warnings"].append(f"会话固定风险: {cb['name']} 登录前后值相同") break # 8. 验证凭证 if verify_url: try: resp = page.goto(verify_url, wait_until="domcontentloaded", timeout=10000) credentials["verification"] = { "url": verify_url, "status": resp.status if resp else None, "title": page.title(), "success": resp.status == 200 if resp else False, } except Exception as e: credentials["verification"] = {"url": verify_url, "error": str(e), "success": False} # 9. 截图 if screenshot_path: try: page.screenshot(path=screenshot_path, full_page=False) print(f" [*] 截图已保存: {screenshot_path}") except Exception as e: print(f" [WARN] 截图失败: {e}", file=sys.stderr) return credentials # ── 单次登录捕获 ─────────────────────────────────────────── def capture_single_login(p, url, verify_url, timeout, role_label, screenshot_dir, role_index, keep_open=False): """单次登录捕获。返回 (credentials, success, browser_obj)""" print(f"\n {'─' * 55}") print(f" │ 角色 {role_index + 1}: {role_label}") print(f" {'─' * 55}") browser = p.chromium.launch( headless=False, args=["--disable-blink-features=AutomationControlled", "--no-first-run", "--no-default-browser-check"], ) context = browser.new_context( ignore_https_errors=True, viewport={"width": 1280, "height": 800}, user_agent=( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36" ), ) page = context.new_page() # 加载目标 URL print(f" [*] 正在加载: {url}") try: page.goto(url, wait_until="networkidle", timeout=60000) except Exception: print(f" [WARN] 加载超时，继续尝试...") # 自动登录页发现 is_login_page = page.evaluate("""() => { if (document.querySelector('input[type="password"]')) return true; let btns = document.querySelectorAll('button, input[type="submit"]'); for (let b of btns) { let t = (b.textContent || b.value || '').toLowerCase(); if (/login|sign.?in|log.?on|登.?录/.test(t)) return true; } return false; }""") if is_login_page: print(f" [+] 当前即为登录页: {page.url}") else: print(f" [!] 当前页面不是登录页: {page.url}") print(f" [*] 正在自动查找登录页...") if any(kw in page.url.lower() for kw in ["/login", "/signin", "/auth"]): print(f" [+] URL 中含 login 关键词，可能已是登录页") else: login_links = page.evaluate("""() => { let links = []; document.querySelectorAll('a').forEach(a => { let text = (a.textContent || '').trim().toLowerCase(); let href = (a.href || '').toLowerCase(); if (/login|sign.?in|登.?录/.test(text) || /[/]login|[/]signin|[/]auth[/]login/.test(href)) { links.push({text: text, href: a.href}); } }); return links; }""") if login_links: try: page.goto(login_links[0]["href"], wait_until="networkidle", timeout=30000) print(f" [+] 已导航到: {page.url}") except Exception: pass else: for try_path in ["/login", "/signin", "/auth/login", "/admin/login"]: try_url = f"{url.rstrip('/')}{try_path}" print(f" [*] 尝试: {try_url}") try: resp = page.goto(try_url, wait_until="networkidle", timeout=15000) has_pwd = page.evaluate("() => !!document.querySelector('input[type=\"password\"]')") if has_pwd: print(f" [+] 找到登录页: {page.url}") break except Exception: continue else: print(f" [*] 未自动发现登录页，将在当前页面等待用户手动导航") print(f" [+] 当前页面: {page.url}") print(f"") print(f" ========================================") print(f" >>> 请在浏览器中完成 [{role_label}] 登录") print(f" >>> 如页面非登录页，请手动导航到登录页") print(f" >>> 等待 {timeout} 秒后自动提取凭证") print(f" ========================================") print(f"") # 登录前 Cookie (会话固定检测) try: cookies_before = context.cookies() cookies_before_list = [{"name": c["name"], "value": c["value"]} for c in (cookies_before or [])] except Exception: cookies_before_list = [] # 监听登录网络请求（响应） captured_logins = [] login_response_success = False def on_response(response): nonlocal login_response_success url_lower = response.url.lower() auth_kw = ["/login", "/auth", "/signin", "/sign-in", "/oauth/token", "/api/login", "/api/auth", "/authenticate", "/access_token", "/token", "/sso", "/session"] is_auth_url = any(kw in url_lower for kw in auth_kw) try: body = response.text() if response.status in (200, 301, 302): preview = body[:200] if is_auth_url: captured_logins.append({"url": response.url, "status": response.status, "body_preview": preview}) # 即使 URL 不含 auth 关键词，响应体含登录成功标志也视为登录成功 if any(kw in body.lower() for kw in ['"success":true', '"code":0', '"code":200', '"result":"ok"', '"status":"ok"', '"access_token"', '"token_type"', '"redirect"', '"login_success"', '"msg":"登录成功"', '"msg":"success"']): login_response_success = True if not is_auth_url: captured_logins.append({"url": response.url, "status": response.status, "body_preview": preview, "detected_by": "body_content"}) except Exception: if is_auth_url and response.status in (200, 301, 302): captured_logins.append({"url": response.url, "status": response.status, "body_preview": "(无法读取)"}) page.on("response", on_response) # 监听用户提交登录表单的动作（请求） login_submitted = False login_submit_time = 0 dom_before_login = None def on_request(request): nonlocal login_submitted, login_submit_time, dom_before_login if login_submitted: return url_lower = request.url.lower() # 扩大检测范围：不仅检查 auth 关键词 URL，也检查所有 POST 中包含登录字段的请求 auth_kw = ["/login", "/auth", "/signin", "/sign-in", "/oauth/token", "/api/login", "/api/auth", "/authenticate", "/access_token", "/token", "/sso", "/session"] is_auth_url = any(kw in url_lower for kw in auth_kw) # 检查 POST body 中是否包含登录字段 has_login_body = False try: post_data = request.post_data if post_data: body_lower = post_data.lower() login_fields = ['username', 'user', 'account', 'email', 'phone', 'mobile', 'passwd', 'pwd', 'captcha', 'verify_code'] if any(f in body_lower for f in login_fields): has_login_body = True except Exception: pass if request.method == "POST" and (is_auth_url or has_login_body): login_submitted = True login_submit_time = time.time() # 在表单提交瞬间捕获 DOM 快照，用于后续对比 try: dom_before_login = page.evaluate("""() => { const body = document.body || document.documentElement; return (body.innerText || '').trim().slice(0, 2000); }""") except Exception: pass label = "auth URL" if is_auth_url else "登录字段 POST" print(f"\n [+] 检测到用户提交登录表单 ({label}) → {request.url}") page.on("request", on_request) initial_url = page.url # 轮询检测登录信号，检测到用户提交表单后自动提取 m, s = divmod(timeout, 60) print(f" [*] 请在浏览器中完成登录，最多等待 {m}分{s}秒...") print(f" [*] 检测到提交登录表单后将立即提取凭证\n") poll_interval = 500 # 每 500ms 检测一次，使用 page.wait_for_timeout 不阻塞事件循环 elapsed = 0 form_submitted = False while elapsed < timeout * 1000: # timeout in ms page.wait_for_timeout(poll_interval) elapsed += poll_interval if login_submitted and not form_submitted: form_submitted = True # 等待 2 秒让 JS 把 token 写入 localStorage page.wait_for_timeout(2000) break # 检测登录信号 signals = get_login_signals(page, initial_url, dom_before_login) for entry in captured_logins: det_label = entry.get("detected_by", "response") signals.append(("网络请求", f"{entry['status']} {entry['url'][:80]}")) if login_submitted: signals.append(("表单提交", "用户已提交登录表单")) if login_response_success: signals.append(("响应确认", "服务端返回登录成功标志")) print(f"\n [*] 检测到的登录信号 ({len(signals)} 个):") for sig_type, sig_val in signals[:10]: print(f" - [{sig_type}] {sig_val}") if not signals: print(f" (无信号 — 尝试提取当前页面凭证)") # 提取凭证 screenshot_path = None if screenshot_dir: safe_label = re.sub(r'[^\w\-]', '_', role_label) screenshot_path = str(Path(screenshot_dir) / f"capture_{role_index + 1}_{safe_label}.png") Path(screenshot_dir).mkdir(parents=True, exist_ok=True) credentials = extract_credentials_from_page( page, cookies_before=cookies_before_list, verify_url=verify_url, screenshot_path=screenshot_path, role_label=role_label, ) # 多维度判定登录是否成功 # 1. 基础凭证检查 (Cookie / Token) has_new_credentials = bool(credentials["cookies"] or credentials["tokens"]) # 2. Cookie 变化检查 (针对"无新cookie但已有cookie生效"的场景) cookies_changed = False if cookies_before_list: before_names = {c["name"] for c in cookies_before_list} after_names = {c["name"] for c in credentials["cookies"]} # 有新 cookie 名称 if after_names - before_names: cookies_changed = True # 或同名 cookie 值变化 else: before_map = {c["name"]: c["value"] for c in cookies_before_list} after_map = {c["name"]: c["value"] for c in credentials["cookies"]} for name in before_names & after_names: if before_map[name] != after_map[name]: cookies_changed = True break # 3. 登录交互检查 (表单提交 / 响应确认) has_login_interaction = login_submitted or login_response_success or bool(captured_logins) # 4. 页面信号检查 (DOM变化 / 表单消失 / Token检测) page_signals = [s for s in signals if s[0].startswith("DOM") or s[0] in ("页面标题", "Token检测", "DOM内容变化", "DOM含登录特征")] # 综合判定 has_auth = False auth_reasons = [] if has_new_credentials and (has_login_interaction or len(signals) >= 2): has_auth = True auth_reasons.append("捕获到凭证") if not has_auth and has_login_interaction and cookies_changed: has_auth = True auth_reasons.append("Cookie变化") if not has_auth and has_login_interaction and len(page_signals) >= 1: has_auth = True auth_reasons.append("页面登录信号") if not has_auth and login_response_success: has_auth = True auth_reasons.append("服务端响应确认") if not has_auth and has_new_credentials and not cookies_before_list: # 登录前没有任何 cookie，登录后有 cookie → 基本可以确定登录成功 has_auth = True auth_reasons.append("登录后出现新凭证") if has_auth: reason_str = " + ".join(auth_reasons) if auth_reasons else "凭证捕获" print(f" [+] 捕获成功 ({reason_str}): {len(credentials['cookies'])} 个 Cookie, " f"{len(credentials['tokens'])} 个 Token") if not cookies_changed and cookies_before_list: print(f" [!] Cookie 值未变化，可能通过已有 Cookie 状态激活登录") else: print(f" [!] 未捕获到认证凭证 (表单提交: {login_submitted}, " f"响应成功: {login_response_success}, " f"信号: {len(signals)}, Cookie变化: {cookies_changed})") if keep_open: print(f" [*] 浏览器保持打开，完成后请手动关闭") else: try: browser.close() except Exception: pass return credentials, has_auth, browser # ── 主入口 ───────────────────────────────────────────────── def interactive_login(url, timeout=120, max_roles=3, verify_url=None, user_data_dir=None, output=None, screenshot=False, keep_open=False): """交互式登录捕获 — 轮询检测登录完成立即提取""" print("\n" + "=" * 65) print(" Vibe Pentest — 交互式浏览器登录凭证捕获") print("=" * 65) print(f" 目标: {url}") print(f" 每角色超时: {timeout} 秒") print(f" 最大角色数: {max_roles}") if verify_url: print(f" 验证 URL: {verify_url}") print("=" * 65) screenshot_dir = None if screenshot: screenshot_dir = Path(output).parent / "screenshots" if output else Path("screenshots") all_roles = [] failures = 0 with sync_playwright() as p: try: for role_idx in range(max_roles): if role_idx == 0: role_label = "admin" this_timeout = timeout else: print(f"\n [?] 已捕获 {len(all_roles)} 个角色") print(f" [?] 还有其他权限的账号吗？(如普通用户/审计员)") print(f" [*] 15 秒内无操作则跳过") time.sleep(15) role_labels = ["admin", "user", "auditor", "readonly"] role_label = role_labels[role_idx] if role_idx < len(role_labels) else f"role_{role_idx + 1}" this_timeout = max(timeout - 30, 45) is_last_role = (role_idx == max_roles - 1) creds, ok, browser = capture_single_login( p, url, verify_url, this_timeout, role_label, screenshot_dir, role_idx, keep_open=(keep_open and is_last_role) ) if keep_open: if ok: all_roles.append(creds) break if ok: # 检查凭证是否与已有角色重复 new_cookies = {c["name"]: c["value"] for c in creds["cookies"]} is_duplicate = False for existing in all_roles: old_cookies = {c["name"]: c["value"] for c in existing.get("cookies", [])} if old_cookies and new_cookies and old_cookies == new_cookies: is_duplicate = True break if not is_duplicate: all_roles.append(creds) print(f" [+] 角色 [{role_label}] 已保存") # 立即保存，防止后续崩溃 if output: output_path = Path(output) output_path.parent.mkdir(parents=True, exist_ok=True) with open(output_path, "w", encoding="utf-8") as f: json.dump({ "target": url, "capture_time": datetime.now(timezone.utc).isoformat(), "roles": all_roles, }, f, indent=2, ensure_ascii=False) print(f" [+] 凭证已立即保存到: {output_path}") else: print(f" [*] 凭证与已有角色相同，跳过") break else: failures += 1 print(f" [!] 角色 [{role_label}] 捕获失败 ({failures}/3)") if failures >= 3: print(f"\n {'─' * 55}") print(f" [!] 3 次登录失败，可能是登录页无法加载或认证方式特殊") print(f" [*] 替代方案：手动提供 Cookie/Token") print(f' AUTH_COOKIE="sessionid=xxx; csrftoken=yyy"') print(f' AUTH_HEADER="Bearer eyJ..."') print(f" {'─' * 55}") break except Exception as e: print(f"\n [WARN] 循环异常: {e}", file=sys.stderr) # 异常时立即保存已捕获的凭证 if output and all_roles: output_path = Path(output) output_path.parent.mkdir(parents=True, exist_ok=True) with open(output_path, "w", encoding="utf-8") as f: json.dump({ "target": url, "capture_time": datetime.now(timezone.utc).isoformat(), "roles": all_roles, }, f, indent=2, ensure_ascii=False) print(f" [+] 异常发生，已保存现有凭证到: {output_path}") # 输出结果 if output: output_path = Path(output) output_path.parent.mkdir(parents=True, exist_ok=True) with open(output_path, "w", encoding="utf-8") as f: json.dump({ "target": url, "capture_time": datetime.now(timezone.utc).isoformat(), "roles": all_roles, }, f, indent=2, ensure_ascii=False) print(f"\n [+] 凭证已保存到: {output_path}") # 格式化摘要 for i, role in enumerate(all_roles): label = role.get("role_label", f"role_{i + 1}") print(f"\n ┌─ 角色 [{label}]") print(f" │ Cookie: {len(role['cookies'])} 个") for c in role["cookies"]: flags = "" if c["httpOnly"]: flags += " [HttpOnly]" if c["secure"]: flags += " [Secure]" val = c["value"][:40] + ("..." if len(c["value"]) > 40 else "") print(f" │ {c['name']} = {val}{flags}") if role.get("tokens"): print(f" │ Token: {len(role['tokens'])} 个") for t in role["tokens"]: val = t["value"][:50] + ("..." if len(t["value"]) > 50 else "") print(f" │ [{t['type']}] {t['key']} = {val}") if role.get("jwt_analysis"): for ja in role["jwt_analysis"]: p = ja.get("payload", {}) print(f" │ JWT: sub={p.get('sub','?')} username={p.get('username','?')} role={p.get('role','?')}") if ja.get("expires_at"): print(f" │ 过期: {ja['expires_at']}") for w in role.get("warnings", []): print(f" │ [!] {w}") if role.get("auth_headers"): for ah in role["auth_headers"]: val = ah["value"][:80] + ("..." if len(ah["value"]) > 80 else "") print(f" │ 认证头: {ah['name']}: {val}") print(f" └{'─' * 40}") return all_roles # ── CDP 模式 ────────────────────────────────────────────── def cdp_login(cdp_port=9222, verify_url=None, output=None, screenshot=False): """通过 CDP 连接已有 Chrome/Edge 实例并捕获凭证""" print(f"[*] 连接 CDP → 127.0.0.1:{cdp_port}") with sync_playwright() as p: try: browser = p.chromium.connect_over_cdp(f"http://127.0.0.1:{cdp_port}") except Exception as e: print(f"[ERROR] 无法连接 CDP: {e}", file=sys.stderr) print(f"[INFO] 请确保 Chrome/Edge 已通过 --remote-debugging-port={cdp_port} 启动", file=sys.stderr) sys.exit(1) contexts = browser.contexts if not contexts: print("[ERROR] 没有找到浏览器上下文", file=sys.stderr) sys.exit(1) context = contexts[0] pages = context.pages if not pages: print("[ERROR] 没有打开的页面", file=sys.stderr) sys.exit(1) page = pages[-1] print(f"[+] 已连接，当前页面: {page.url}") print("[*] 正在提取认证凭证...") try: cookies_before = context.cookies() cookies_before_list = [{"name": c["name"], "value": c["value"]} for c in (cookies_before or [])] except Exception: cookies_before_list = [] screenshot_path = None if screenshot and output: ss_dir = Path(output).parent / "screenshots" ss_dir.mkdir(parents=True, exist_ok=True) screenshot_path = str(ss_dir / "cdp_capture.png") credentials = extract_credentials_from_page( page, cookies_before=cookies_before_list, verify_url=verify_url, screenshot_path=screenshot_path, role_label="cdp_capture", ) browser.close() roles = [credentials] if output: output_path = Path(output) output_path.parent.mkdir(parents=True, exist_ok=True) with open(output_path, "w", encoding="utf-8") as f: json.dump({ "capture_time": datetime.now(timezone.utc).isoformat(), "roles": roles, }, f, indent=2, ensure_ascii=False) print(f"\n[+] 凭证已保存到: {output_path}") for role in roles: print(f"\n ┌─ 角色 [{role.get('role_label', 'cdp')}]") print(f" │ Cookie: {len(role['cookies'])} 个") for c in role["cookies"]: val = c["value"][:40] + ("..." if len(c["value"]) > 40 else "") print(f" │ {c['name']} = {val}") if role.get("auth_headers"): for ah in role["auth_headers"]: val = ah["value"][:80] + ("..." if len(ah["value"]) > 80 else "") print(f" │ 认证头: {ah['name']}: {val}") print(f" └{'─' * 40}") return roles # ── CLI ──────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser( description="Vibe Pentest — 交互式浏览器登录凭证捕获", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 示例: python extract_credentials.py --url https://target.com/login python extract_credentials.py --url https://target.com --timeout 120 --username admin --output sessions/admin.json python extract_credentials.py --cdp-port 9222 --output sessions/admin.json """, ) parser.add_argument("--url", help="登录页面 URL") parser.add_argument("--timeout", type=int, default=120, help="每个角色的等待超时（秒），默认 120") parser.add_argument("--roles", type=int, default=3, help="最大角色数，默认 3") parser.add_argument("--verify-url", help="登录完成后验证凭证的目标 URL") parser.add_argument("--output", "-o", help="输出文件路径（JSON 格式）") parser.add_argument("--screenshot", "-s", action="store_true", help="保存页面截图") parser.add_argument("--cdp-port", type=int, help="CDP 端口号（连接已有 Chrome/Edge）") parser.add_argument("--user-data-dir", help="Chrome 用户数据目录（保留登录状态）") parser.add_argument("--keep-open", "-k", action="store_true", help="凭证提取后不关闭浏览器") parser.add_argument("--username", help="用户名标签（已废弃，使用 --roles 替代）") args = parser.parse_args() if args.cdp_port: cdp_login(args.cdp_port, args.verify_url, args.output, args.screenshot) elif args.url: interactive_login( url=args.url, timeout=args.timeout, max_roles=args.roles, verify_url=args.verify_url, user_data_dir=args.user_data_dir, output=args.output, screenshot=args.screenshot, keep_open=args.keep_open, ) else: parser.error("需要指定 --url 或 --cdp-port") if __name__ == "__main__": main()