1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126
| import asyncio, json, subprocess, time, urllib.request, base64, os, signal, sys, websockets
CMD_ID = [0] async def cdp(ws, method, params=None, session_id=None): CMD_ID[0] += 1; msg = {"id": CMD_ID[0], "method": method, "params": params or {}} if session_id: msg["sessionId"] = session_id await ws.send(json.dumps(msg)) async for resp in ws: data = json.loads(resp) if data.get("id") == CMD_ID[0]: return data.get("result", {})
class CloakBrowserManager: """Complete CloakBrowser manager: launch, pool, fingerprint, cleanup.""" PATHS = {"win32": [r"C:\Users\Lenovo\.cloakbrowser\chromium-146.0.7680.177.5\chrome.exe"], "darwin": ["/Applications/CloakBrowser.app/Contents/MacOS/CloakBrowser"], "linux": ["/opt/cloakbrowser/chrome"]} DEFAULT_FP = {"viewport": {"width":1920,"height":1080,"scale":1}, "timezone":"Asia/Shanghai","locale":"zh-CN", "webgl_vendor":"Intel Inc.", "webgl_renderer":"Intel Iris", "hardware_concurrency":8, "device_memory":4} def __init__(self, chrome_path=None, base_port=9222): self.chrome = chrome_path or self._find() self.instances = {}; self._next = base_port def _find(self): for p in self.PATHS.get(sys.platform, []): if os.path.exists(p): return p raise FileNotFoundError("CloakBrowser not found") async def spawn(self, name=None, fp=None): port = self._next; self._next += 1 name = name or f"p{port}" dp = os.path.join(os.path.dirname(self.chrome), "profiles", name) os.makedirs(dp, exist_ok=True) proc = subprocess.Popen([ self.chrome, f"--remote-debugging-port={port}", "--remote-allow-origins=*","--no-first-run","--no-default-browser-check", f"--user-data-dir={dp}","--disable-sync","--disable-background-networking", ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) time.sleep(2) ws = await websockets.connect( json.loads(urllib.request.urlopen( f"http://127.0.0.1:{port}/json/version").read())["webSocketDebuggerUrl"], max_size=2**24) await cdp(ws, "Page.enable"); await cdp(ws, "Runtime.enable") await self._apply_fp(ws, fp or self.DEFAULT_FP) self.instances[port] = {"proc": proc, "ws": ws} print(f"[Mgr] Spawned {port} ({name})") return ws, port async def _apply_fp(self, ws, fp): vp = fp.get("viewport", {}) await cdp(ws, "Emulation.setDeviceMetricsOverride", {"width":vp.get("width",1920),"height":vp.get("height",1080), "deviceScaleFactor":vp.get("scale",1),"mobile":False, "screenWidth":vp.get("width",1920),"screenHeight":vp.get("height",1080)}) if "timezone" in fp: await cdp(ws, "Emulation.setTimezoneOverride", {"timezoneId":fp["timezone"]}) if "user_agent" in fp: await cdp(ws, "Emulation.setUserAgentOverride", {"userAgent":fp["user_agent"],"platform":fp.get("platform","Win32")}) loc = fp.get("locale","zh-CN") await cdp(ws, "Page.addScriptToEvaluateOnNewDocument", {"source":f""" Object.defineProperty(navigator,'webdriver',{{get:()=>undefined}}); Object.defineProperties(navigator,{{language:{{get:()=>'{loc}'}}, languages:{{get:()=>['{loc}','en','en-US']}}, hardwareConcurrency:{{get:()=>{fp.get('hardware_concurrency',8)}}}, deviceMemory:{{get:()=>{fp.get('device_memory',4)}}}}}); if(WebGLRenderingContext){{const _p=WebGLRenderingContext.prototype.getParameter; WebGLRenderingContext.prototype.getParameter=function(p){{ if(p===0x9245)return'{fp.get("webgl_vendor","Intel Inc.")}'; if(p===0x9246)return'{fp.get("webgl_renderer","Intel Iris")}'; if(p===0x1F00)return'WebKit';if(p===0x1F01)return'WebKit WebGL'; return _p.call(this,p);}};}} if(navigator.plugins.length===0){{const p=[ {{name:'Chrome PDF Plugin',filename:'internal-pdf-viewer'}}, {{name:'Chrome PDF Viewer',filename:'mhjfbmdgcfjbbpaeojofohoefgiehjai'}}, {{name:'Native Client',filename:'internal-nacl-plugin'}}]; p.item=i=>p[i];p.namedItem=n=>p.find(x=>x.name===n); Object.defineProperty(navigator,'plugins',{{get:()=>p}});}} """}) async def snap(self, ws, path): r = await cdp(ws, "Page.captureScreenshot", {"format":"png"}) with open(path, "wb") as f: f.write(base64.b64decode(r["data"])) async def kill(self, port=None): targets = [port] if port else list(self.instances.keys()) for p in targets: inst = self.instances.pop(p, None) if not inst: continue try: await inst["ws"].close() except: pass if inst["proc"]: try: inst["proc"].terminate(); inst["proc"].wait(timeout=5) except: inst["proc"].kill(); inst["proc"].wait() print(f"[Mgr] Killed {p}") @property def count(self): return len(self.instances) async def __aenter__(self): return self async def __aexit__(self, *a): await self.kill()
async def demo(): async with CloakBrowserManager() as mgr: w1, _ = await mgr.spawn("demo-1", { "viewport":{"width":1920,"height":1080},"timezone":"Asia/Shanghai", "locale":"zh-CN","platform":"Win32","webgl_vendor":"Intel Inc.", "user_agent":"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36"}) w2, _ = await mgr.spawn("demo-2", { "viewport":{"width":1440,"height":900,"scale":2},"timezone":"America/New_York", "locale":"en-US","platform":"MacIntel","webgl_vendor":"Apple Inc.", "user_agent":"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36"}) print(f"Active: {mgr.count}") for w, u, fn in [(w1,"https://www.baidu.com","baidu.png"), (w2,"https://www.google.com","google.png")]: await cdp(w, "Page.navigate", {"url": u}); await asyncio.sleep(2) await mgr.snap(w, fn)
asyncio.run(demo())
|