From 5d4f5dcac1dbafd344e7fe5d7cf50024c3ccfbed Mon Sep 17 00:00:00 2001 From: liubo235 Date: Thu, 18 Dec 2025 14:33:18 +0800 Subject: [PATCH] feat: Enhance anti-fraud detection and fix local testing - Anti-Fraud Enhancements: - Add realistic browser fingerprinting (WebGL, Canvas noise) - Improve User-Agent and HTTP headers configuration - Add human behavior simulation (random delays, mouse movement) - Create comprehensive anti-fraud guide (docs/anti_fraud_guide.md) - Bug Fixes: - Fix Gateway routing loop (127.0.0.1:8000) by using container names in Docker - Improve NodeRegistry IP detection for Docker/local environments - Fix missing asyncio import in gateway.py WebSocket proxy - Fix input field keyboard issues in remote.html (Backspace, Arrow keys) - Local Testing: - Add one-command local test script (scripts/start_local_test.py) - Add gateway API test script (scripts/test_gateway_local.py) - Update README.md with local testing instructions - Improve error messages for API Key authentication - Docker: - Add NODE_HOST environment variable for proper container networking - Update docker-compose.yml with container name configuration --- README.md | 82 +++++++++++++ cdp_client.py | 124 +++++++++++++++++++- core/registry.py | 27 ++++- docker-compose.yml | 2 + docs/anti_fraud_guide.md | 174 ++++++++++++++++++++++++++++ gateway.py | 10 +- scripts/start_local_test.py | 211 ++++++++++++++++++++++++++++++++++ scripts/test_gateway_local.py | 84 ++++++++++++++ 8 files changed, 703 insertions(+), 11 deletions(-) create mode 100644 docs/anti_fraud_guide.md create mode 100755 scripts/start_local_test.py create mode 100644 scripts/test_gateway_local.py diff --git a/README.md b/README.md index ec1c179..b575bf7 100644 --- a/README.md +++ b/README.md @@ -35,6 +35,88 @@ http://localhost:8000/static/remote.html > Note: By default, `API_KEY` is set to `dev-test-key`. You might need to configure headers if using the UI directly, or disable auth in `docker-compose.yml` for testing. +## 🧪 Local Testing (Non-Docker) + +For local development and testing without Docker, you can start all services manually. + +### Quick Start (One Command) + +```bash +python scripts/start_local_test.py +``` + +This script will: +- ✅ Check and start Redis (if not running) +- ✅ Start 2 Worker nodes (ports 8001, 8002) +- ✅ Start Gateway (port 8000) +- ✅ Run automatic API tests +- ✅ Display access URLs + +Press `Ctrl+C` to stop all services. + +### Manual Setup (Step by Step) + +**1. Start Redis** (if not installed): + +```bash +# macOS +brew install redis +brew services start redis + +# Linux +sudo apt-get install redis-server +sudo systemctl start redis +``` + +**2. Start Worker Node 1** (Terminal 1): + +```bash +export HTTP_PORT=8001 +export MAX_SESSIONS=5 +export REDIS_URL=redis://localhost:6379/0 +export CLUSTER_SECRET=dev-cluster-secret +export NODE_HOST=localhost +python http_server.py +``` + +**3. Start Worker Node 2** (Terminal 2): + +```bash +export HTTP_PORT=8002 +export MAX_SESSIONS=5 +export REDIS_URL=redis://localhost:6379/0 +export CLUSTER_SECRET=dev-cluster-secret +export NODE_HOST=localhost +python http_server.py +``` + +**4. Start Gateway** (Terminal 3): + +```bash +export REDIS_URL=redis://localhost:6379/0 +export API_KEY=dev-test-key +export CLUSTER_SECRET=dev-cluster-secret +python gateway.py +``` + +**5. Test the Setup**: + +```bash +python scripts/test_gateway_local.py +``` + +### Access URLs + +- **Gateway**: http://localhost:8000 +- **Remote Control UI**: http://localhost:8000/static/remote.html +- **API Documentation**: http://localhost:8000/docs + +### Notes + +- For local testing, `NODE_HOST` should be set to `localhost` (not container names) +- Ensure Redis is running before starting services +- The script will automatically check Redis availability + ## 📦 Installation (Local Development) ### 1. Install Dependencies diff --git a/cdp_client.py b/cdp_client.py index 5dd1a20..88a5355 100644 --- a/cdp_client.py +++ b/cdp_client.py @@ -153,15 +153,41 @@ async def initialize( # 启动浏览器 self.browser = await self.playwright.chromium.launch(**launch_options) - # 创建上下文 + # 创建上下文（增强反检测配置） + # 使用真实的 Chrome User-Agent（如果没有提供） + default_user_agent = user_agent or ( + 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ' + '(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36' + ) + context_options = { 'viewport': {'width': width, 'height': height}, 'bypass_csp': True, + 'user_agent': default_user_agent, + # 设置更真实的 HTTP headers + 'extra_http_headers': { + 'Accept-Language': 'zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7', + 'Accept-Encoding': 'gzip, deflate, br', + 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8', + 'Connection': 'keep-alive', + 'Upgrade-Insecure-Requests': '1', + 'Sec-Fetch-Dest': 'document', + 'Sec-Fetch-Mode': 'navigate', + 'Sec-Fetch-Site': 'none', + 'Sec-Fetch-User': '?1', + 'Cache-Control': 'max-age=0', + }, + # 设置更真实的屏幕参数 + 'screen': { + 'width': width, + 'height': height + }, + # 设置时区（中国时区） + 'timezone_id': 'Asia/Shanghai', + # 设置语言 + 'locale': 'zh-CN', } - if user_agent: - context_options['user_agent'] = user_agent - if proxy: context_options['proxy'] = proxy @@ -196,6 +222,69 @@ async def initialize( # 应用 playwright-stealth 的完整反检测补丁 await stealth_async(self.page) + # 增强反检测：额外的指纹伪装 + await self.page.add_init_script(""" + // Canvas 指纹随机化（添加噪声） + const originalToDataURL = HTMLCanvasElement.prototype.toDataURL; + HTMLCanvasElement.prototype.toDataURL = function(type) { + const context = this.getContext('2d'); + if (context) { + const imageData = context.getImageData(0, 0, this.width, this.height); + for (let i = 0; i < imageData.data.length; i += 4) { + // 添加微小的随机噪声（不影响视觉） + imageData.data[i] += Math.random() * 0.01; + } + context.putImageData(imageData, 0, 0); + } + return originalToDataURL.apply(this, arguments); + }; + + // WebGL 指纹伪装 + const getParameter = WebGLRenderingContext.prototype.getParameter; + WebGLRenderingContext.prototype.getParameter = function(parameter) { + if (parameter === 37445) { // UNMASKED_VENDOR_WEBGL + return 'Intel Inc.'; + } + if (parameter === 37446) { // UNMASKED_RENDERER_WEBGL + return 'Intel Iris OpenGL Engine'; + } + return getParameter.apply(this, arguments); + }; + + // 完善 navigator 属性 + Object.defineProperty(navigator, 'hardwareConcurrency', { + get: () => 8 + }); + Object.defineProperty(navigator, 'deviceMemory', { + get: () => 8 + }); + Object.defineProperty(navigator, 'platform', { + get: () => 'Win32' + }); + + // 设置语言 + Object.defineProperty(navigator, 'language', { + get: () => 'zh-CN' + }); + Object.defineProperty(navigator, 'languages', { + get: () => ['zh-CN', 'zh', 'en-US', 'en'] + }); + + // 完善 screen 对象 + Object.defineProperty(screen, 'availWidth', { + get: () => window.innerWidth || 1920 + }); + Object.defineProperty(screen, 'availHeight', { + get: () => window.innerHeight || 1080 + }); + Object.defineProperty(screen, 'colorDepth', { + get: () => 24 + }); + Object.defineProperty(screen, 'pixelDepth', { + get: () => 24 + }); + """) + # 调试 webdriver 状态 info = await self.page.evaluate(""" () => ({ @@ -220,12 +309,35 @@ async def initialize( raise async def navigate(self, url: str, timeout: int = 30) -> str: - """导航到指定 URL""" + """导航到指定 URL（增强反检测：添加人类行为模拟）""" if not self.page: raise RuntimeError("浏览器会话未初始化") try: - response = await self.page.goto(url, timeout=timeout * 1000, wait_until='domcontentloaded') + # 添加随机延迟，模拟人类行为（0.5-2秒） + import random + delay = random.uniform(0.5, 2.0) + await asyncio.sleep(delay) + + # 导航时设置更真实的 referer（如果是首次访问则没有 referer） + response = await self.page.goto( + url, + timeout=timeout * 1000, + wait_until='domcontentloaded', + referer=None # 首次访问不设置 referer + ) + + # 页面加载后，模拟一些人类行为（随机鼠标移动） + try: + await asyncio.sleep(random.uniform(0.3, 1.0)) + # 随机移动鼠标到页面中心附近 + await self.page.mouse.move( + random.randint(400, 800), + random.randint(300, 600) + ) + except Exception as e: + logger.debug(f"模拟鼠标移动失败（可忽略）: {e}") + self.last_activity = time.time() return self.page.url except Exception as e: diff --git a/core/registry.py b/core/registry.py index a407220..cd9d74e 100644 --- a/core/registry.py +++ b/core/registry.py @@ -24,11 +24,32 @@ def __init__(self, redis_url: str): self.is_running = False def _get_local_ip(self): + """获取本机 IP 或容器名（Docker 环境）""" + # 优先使用环境变量（Docker Compose 可以设置） + node_host = os.getenv('NODE_HOST') + if node_host: + return node_host + + # 在 Docker 环境中，尝试使用容器名（hostname） + # 这样 Gateway 可以通过容器名访问 Worker try: - # 获取本机IP，在Docker中这通常是容器IP - # 在某些网络配置下可能需要手动指定 hostname = socket.gethostname() - return socket.gethostbyname(hostname) + # 检查是否在 Docker 中（hostname 通常是容器名） + # 如果 hostname 不是 localhost 或 127.0.0.1，很可能是容器名 + if hostname and hostname not in ('localhost', '127.0.0.1'): + # 尝试解析，如果失败则直接使用 hostname（容器名） + try: + ip = socket.gethostbyname(hostname) + # 如果解析出的是 127.0.0.1，说明不在 Docker 网络中，使用 hostname + if ip == '127.0.0.1': + return hostname + return ip + except: + # 解析失败，直接使用 hostname（容器名） + return hostname + else: + # 非 Docker 环境，使用 IP + return socket.gethostbyname(hostname) if hostname else '127.0.0.1' except: return '127.0.0.1' diff --git a/docker-compose.yml b/docker-compose.yml index 4033bdf..9778241 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -27,6 +27,7 @@ services: - MAX_SESSIONS=5 - REDIS_URL=redis://redis:6379/0 - CLUSTER_SECRET=dev-cluster-secret + - NODE_HOST=browser-node-1 # 使用容器名，Gateway 可以通过此名称访问 volumes: - ./log:/app/log depends_on: @@ -41,6 +42,7 @@ services: - MAX_SESSIONS=5 - REDIS_URL=redis://redis:6379/0 - CLUSTER_SECRET=dev-cluster-secret + - NODE_HOST=browser-node-2 # 使用容器名，Gateway 可以通过此名称访问 volumes: - ./log:/app/log depends_on: diff --git a/docs/anti_fraud_guide.md b/docs/anti_fraud_guide.md new file mode 100644 index 0000000..98067d5 --- /dev/null +++ b/docs/anti_fraud_guide.md @@ -0,0 +1,174 @@ +# 反风控策略优化指南 + +## 当前已实现的增强功能 + +### 1. 浏览器指纹伪装 +- ✅ 真实的 Chrome User-Agent +- ✅ 完整的 HTTP Headers（Accept-Language, Accept-Encoding 等） +- ✅ WebGL 指纹伪装（Intel Inc. / Intel Iris） +- ✅ Canvas 指纹随机化（添加噪声） +- ✅ Screen 参数完善 +- ✅ Navigator 属性完善（hardwareConcurrency, deviceMemory 等） +- ✅ 时区和语言设置（Asia/Shanghai, zh-CN） + +### 2. 行为模拟 +- ✅ 随机延迟（0.5-2秒）在导航前 +- ✅ 页面加载后随机鼠标移动 +- ✅ playwright-stealth 完整补丁 + +## 针对严格风控（如小红书）的进一步优化建议 + +### 方案 1: 使用住宅代理 IP（推荐） + +**问题**：数据中心 IP（如 AWS、阿里云）容易被标记。 + +**解决方案**： +```python +# 在创建会话时传入代理 +client = BrowserHTTPClient(base_url='http://localhost:8000') +await client.create_session( + headless=True, + proxy={'server': 'http://your-residential-proxy:port'} +) +``` + +**推荐代理服务**： +- 住宅代理：Bright Data, Smartproxy, Oxylabs +- 移动代理：对于移动端检测更有效 + +### 方案 2: 添加 Cookies 和 Session + +**问题**：首次访问容易被标记为新用户。 + +**解决方案**： +```python +# 在导航前设置 cookies +await client.set_cookies([ + { + 'name': 'session_id', + 'value': 'your_session_value', + 'domain': '.xiaohongshu.com', + 'path': '/', + 'secure': True, + 'httpOnly': True + } +]) +``` + +**获取 Cookies 方法**： +1. 手动登录一次，导出 cookies +2. 使用浏览器插件（如 EditThisCookie）导出 +3. 通过正常浏览器访问后，使用 DevTools 复制 cookies + +### 方案 3: 使用移动端 User-Agent + +**问题**：某些网站对移动端检测更宽松。 + +**解决方案**： +```python +# 使用移动端 UA +mobile_ua = ( + 'Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) ' + 'AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.0 Mobile/15E148 Safari/604.1' +) + +await client.create_session( + headless=True, + user_agent=mobile_ua, + width=375, # iPhone 宽度 + height=812 # iPhone 高度 +) +``` + +### 方案 4: 增加访问预热 + +**问题**：直接访问目标网站容易被识别。 + +**解决方案**： +```python +# 先访问一些常见网站，建立"正常"访问历史 +warmup_urls = [ + 'https://www.baidu.com', + 'https://www.google.com', + 'https://www.zhihu.com' +] + +for url in warmup_urls: + await client.navigate(url) + await asyncio.sleep(random.uniform(2, 5)) # 模拟阅读时间 + +# 然后再访问目标网站 +await client.navigate('https://www.xiaohongshu.com') +``` + +### 方案 5: 降低访问频率 + +**问题**：高频访问容易被限流。 + +**解决方案**： +- 在请求之间添加随机延迟（5-30秒） +- 使用队列系统，控制并发数 +- 实现指数退避重试策略 + +### 方案 6: 使用真实浏览器环境（非 Headless） + +**问题**：Headless 模式更容易被检测。 + +**解决方案**： +```python +# 设置 headless=False（需要 Xvfb 在服务器上） +await client.create_session(headless=False) +``` + +**注意**：在 Docker 中需要安装 Xvfb： +```dockerfile +RUN apt-get install -y xvfb +ENV DISPLAY=:99 +CMD ["Xvfb", ":99", "-screen", "0", "1920x1080x24"] & +``` + +## 检测和调试 + +### 检查当前指纹 +```python +fingerprint = await client.execute_script(""" + return { + userAgent: navigator.userAgent, + platform: navigator.platform, + language: navigator.language, + languages: navigator.languages, + webdriver: navigator.webdriver, + plugins: Array.from(navigator.plugins).map(p => p.name), + screen: { + width: screen.width, + height: screen.height, + colorDepth: screen.colorDepth + }, + hardwareConcurrency: navigator.hardwareConcurrency, + deviceMemory: navigator.deviceMemory + }; +""") +print(json.dumps(fingerprint, indent=2)) +``` + +### 测试反检测效果 +访问以下检测网站： +- https://bot.sannysoft.com/ +- https://arh.antoinevastel.com/bots/areyouheadless +- https://pixelscan.net/ + +## 最佳实践总结 + +1. **组合使用**：代理 + Cookies + 预热 + 延迟 +2. **环境隔离**：每个任务使用独立的浏览器会话 +3. **监控告警**：检测到风控时自动切换策略 +4. **成本平衡**：住宅代理成本较高，根据需求选择 + +## 紧急应对措施 + +如果遇到风控： +1. **立即停止**：避免 IP 被永久封禁 +2. **更换代理**：使用新的住宅 IP +3. **清理痕迹**：删除 cookies，更换 User-Agent +4. **降低频率**：增加延迟，减少并发 + diff --git a/gateway.py b/gateway.py index 1b4f684..c9cf197 100644 --- a/gateway.py +++ b/gateway.py @@ -45,7 +45,9 @@ async def verify_client_auth( api_key: str = Security(api_key_header) ): """验证客户端 API Key (支持 Header 和 Query Param)""" + # 如果未配置 API_KEY，则跳过验证（本地开发模式） if not config.API_KEY: + logger.debug("API_KEY not configured, skipping authentication") return # 1. 优先检查 Header (由 Security 自动提取，若无则 api_key 为 None) @@ -57,8 +59,12 @@ async def verify_client_auth( if query_token == config.API_KEY: return - # 3. 验证失败 - raise HTTPException(status_code=403, detail="Invalid API Key") + # 3. 验证失败 - 提供更详细的错误信息 + logger.warning(f"API Key validation failed. Header: {api_key}, Query: {query_token}, Expected: {config.API_KEY}") + raise HTTPException( + status_code=403, + detail=f"Invalid API Key. Please provide 'X-API-Key' header or 'token' query parameter. Expected: {config.API_KEY}" + ) def get_upstream_headers(original_headers: dict = None) -> dict: """构造上游请求头，注入内部通信密钥""" diff --git a/scripts/start_local_test.py b/scripts/start_local_test.py new file mode 100755 index 0000000..95370a1 --- /dev/null +++ b/scripts/start_local_test.py @@ -0,0 +1,211 @@ +#!/usr/bin/env python3 +""" +本地测试启动脚本 +用于在非 Docker 环境中快速启动 Gateway + Worker + Redis +""" +import subprocess +import sys +import os +import time +import signal +import requests + +processes = [] + +def cleanup(): + """清理进程""" + print("\n清理进程...") + for p in processes: + try: + p.terminate() + p.wait(timeout=5) + except: + try: + p.kill() + except: + pass + print("清理完成") + +def check_redis(): + """检查 Redis 是否运行""" + try: + import redis + r = redis.Redis(host='localhost', port=6379, decode_responses=True) + r.ping() + return True + except: + return False + +def start_redis(): + """启动 Redis（如果未运行）""" + if check_redis(): + print("✅ Redis 已在运行") + return None + + print("启动 Redis...") + # 尝试使用本地 redis-server + try: + p = subprocess.Popen( + ['redis-server', '--port', '6379'], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE + ) + time.sleep(2) + if check_redis(): + print("✅ Redis 启动成功") + return p + else: + print("❌ Redis 启动失败，请手动启动: redis-server") + return None + except FileNotFoundError: + print("⚠️ redis-server 未找到，请手动启动 Redis:") + print(" macOS: brew install redis && brew services start redis") + print(" Linux: sudo apt-get install redis-server && sudo systemctl start redis") + return None + +def start_worker(port, node_name): + """启动 Worker 节点""" + env = os.environ.copy() + env.update({ + 'HTTP_PORT': str(port), + 'MAX_SESSIONS': '5', + 'REDIS_URL': 'redis://localhost:6379/0', + 'CLUSTER_SECRET': 'dev-cluster-secret', + 'NODE_HOST': 'localhost' # 本地环境使用 localhost + }) + + print(f"启动 Worker {node_name} (端口 {port})...") + p = subprocess.Popen( + [sys.executable, 'http_server.py'], + env=env, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT + ) + return p + +def start_gateway(): + """启动 Gateway""" + env = os.environ.copy() + env.update({ + 'REDIS_URL': 'redis://localhost:6379/0', + 'API_KEY': 'dev-test-key', + 'CLUSTER_SECRET': 'dev-cluster-secret' + }) + + print("启动 Gateway (端口 8000)...") + p = subprocess.Popen( + [sys.executable, 'gateway.py'], + env=env, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT + ) + return p + +def wait_for_service(url, name, timeout=30): + """等待服务启动""" + import time + start = time.time() + while time.time() - start < timeout: + try: + resp = requests.get(url, timeout=2) + if resp.status_code < 500: + print(f"✅ {name} 已就绪") + return True + except: + pass + time.sleep(1) + print(f"❌ {name} 启动超时") + return False + +def test_gateway(): + """测试 Gateway""" + print("\n" + "="*50) + print("测试 Gateway API...") + print("="*50) + + headers = { + "X-API-Key": "dev-test-key", + "Content-Type": "application/json" + } + + try: + resp = requests.post( + "http://localhost:8000/api/sessions", + json={"headless": True}, + headers=headers, + timeout=10 + ) + + if resp.status_code == 200: + data = resp.json() + print(f"✅ 创建会话成功: {data.get('session_id')}") + return True + else: + print(f"❌ 创建会话失败: {resp.status_code} - {resp.text}") + return False + except Exception as e: + print(f"❌ 测试失败: {e}") + return False + +def main(): + print("="*50) + print("Browser RPC - 本地测试环境启动") + print("="*50) + + # 注册清理函数 + signal.signal(signal.SIGINT, lambda s, f: cleanup() or sys.exit(0)) + signal.signal(signal.SIGTERM, lambda s, f: cleanup() or sys.exit(0)) + + # 1. 启动 Redis + redis_proc = start_redis() + if redis_proc: + processes.append(redis_proc) + + if not check_redis(): + print("❌ Redis 未运行，无法继续") + print("请先启动 Redis，然后重新运行此脚本") + sys.exit(1) + + # 2. 启动 Worker 节点 + worker1 = start_worker(8001, "Worker-1") + processes.append(worker1) + + worker2 = start_worker(8002, "Worker-2") + processes.append(worker2) + + # 3. 启动 Gateway + gateway = start_gateway() + processes.append(gateway) + + # 4. 等待服务启动 + print("\n等待服务启动...") + time.sleep(5) + + wait_for_service("http://localhost:8001/", "Worker-1") + wait_for_service("http://localhost:8002/", "Worker-2") + wait_for_service("http://localhost:8000/", "Gateway") + + # 5. 测试 + print("\n" + "="*50) + print("服务启动完成！") + print("="*50) + print("\n访问地址:") + print(" - Gateway: http://localhost:8000") + print(" - 远程控制: http://localhost:8000/static/remote.html") + print(" - API 文档: http://localhost:8000/docs") + print("\n按 Ctrl+C 停止所有服务\n") + + # 运行测试 + time.sleep(3) + test_gateway() + + # 保持运行 + try: + while True: + time.sleep(1) + except KeyboardInterrupt: + cleanup() + +if __name__ == "__main__": + main() + diff --git a/scripts/test_gateway_local.py b/scripts/test_gateway_local.py new file mode 100644 index 0000000..836e36e --- /dev/null +++ b/scripts/test_gateway_local.py @@ -0,0 +1,84 @@ +#!/usr/bin/env python3 +""" +本地测试脚本 - 演示如何正确调用 Gateway API +""" +import requests +import sys + +# 配置 +GATEWAY_URL = "http://localhost:8000" +API_KEY = "dev-test-key" # 对应 docker-compose.yml 中的配置 + +def test_create_session(): + """测试创建会话""" + print(f"Testing: POST {GATEWAY_URL}/api/sessions") + print(f"API Key: {API_KEY}") + + headers = { + "X-API-Key": API_KEY, + "Content-Type": "application/json" + } + + try: + resp = requests.post( + f"{GATEWAY_URL}/api/sessions", + json={"headless": True, "width": 1280, "height": 720}, + headers=headers + ) + + print(f"Status Code: {resp.status_code}") + + if resp.status_code == 200: + data = resp.json() + print(f"✅ Success! Session ID: {data.get('session_id')}") + return data.get('session_id') + else: + print(f"❌ Failed: {resp.text}") + return None + + except requests.exceptions.ConnectionError: + print(f"❌ Connection Error: Cannot connect to {GATEWAY_URL}") + print(" Make sure Gateway is running: docker compose up -d gateway") + return None + except Exception as e: + print(f"❌ Error: {e}") + return None + +def test_without_auth(): + """测试不带 API Key 的情况（应该失败）""" + print("\n" + "="*50) + print("Testing without API Key (should fail):") + print("="*50) + + try: + resp = requests.post( + f"{GATEWAY_URL}/api/sessions", + json={"headless": True} + ) + print(f"Status Code: {resp.status_code}") + print(f"Response: {resp.text}") + except Exception as e: + print(f"Error: {e}") + +if __name__ == "__main__": + print("="*50) + print("Browser RPC Gateway - Local Test") + print("="*50) + + # 测试带认证的请求 + session_id = test_create_session() + + # 测试不带认证的请求（演示错误） + if len(sys.argv) > 1 and sys.argv[1] == "--show-error": + test_without_auth() + + if session_id: + print(f"\n✅ Test passed! You can now use session_id: {session_id}") + print(f"\nNext steps:") + print(f" 1. Navigate: POST {GATEWAY_URL}/api/sessions/{session_id}/navigate") + print(f" 2. Screenshot: POST {GATEWAY_URL}/api/sessions/{session_id}/screenshot") + print(f" 3. Close: DELETE {GATEWAY_URL}/api/sessions/{session_id}") + else: + print("\n❌ Test failed. Check the error messages above.") + sys.exit(1) +