272 lines
12 KiB
Python
272 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
# COOKIE VAULT v5 — session-1 chrome launcher bridge
|
|
# Runs AS drjones (interactive Session 1) via scheduled task.
|
|
# Listens 127.0.0.1:5067, receives {cookies, domain} JSON, launches visible Chrome+selenium.
|
|
import base64
|
|
import json, os, re, time, threading
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
from urllib.parse import urlparse
|
|
|
|
SESSION_DIR = r"C:\cookievault\sessions"
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def log_message(self, *a): pass
|
|
def do_GET(self):
|
|
if self.path.startswith("/shot/"):
|
|
tok = self.path[len("/shot/"):]
|
|
img = SCREENSHOTS.get(tok)
|
|
if not img:
|
|
self.send_response(404); self.end_headers(); return
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "image/png")
|
|
self.send_header("Content-Length", str(len(img)))
|
|
self.end_headers()
|
|
self.wfile.write(img)
|
|
return
|
|
self.send_response(404); self.end_headers()
|
|
def do_POST(self):
|
|
if urlparse(self.path).path == "/check":
|
|
try:
|
|
ln = int(self.headers.get("Content-Length", 0))
|
|
data = json.loads(self.rfile.read(ln))
|
|
domain = data.get("domain", "")
|
|
cookies = data.get("cookies", [])
|
|
proxy = data.get("proxy", "")
|
|
if not domain or not cookies:
|
|
self._json(400, {"ok": False, "error": "domain and cookies required"}); return
|
|
res = check_login(domain, cookies, proxy)
|
|
self._json(200, res)
|
|
except Exception as e:
|
|
self._json(500, {"ok": False, "error": str(e)})
|
|
return
|
|
if urlparse(self.path).path != "/launch":
|
|
self.send_response(404); self.end_headers(); return
|
|
try:
|
|
ln = int(self.headers.get("Content-Length", 0))
|
|
data = json.loads(self.rfile.read(ln))
|
|
domain = data.get("domain", "")
|
|
cookies = data.get("cookies", [])
|
|
proxy = data.get("proxy", "") # e.g. "http://10.30.20.154:3128"
|
|
if not domain or not cookies:
|
|
self._json(400, {"ok": False, "error": "domain and cookies required"}); return
|
|
tok = "%s_%d" % (re.sub(r'[^a-z0-9]', '', domain.lower())[:20], int(time.time()))
|
|
threading.Thread(target=launch, args=(domain, cookies, tok, proxy), daemon=True).start()
|
|
self._json(200, {"ok": True, "launched": len(cookies), "token": tok})
|
|
except Exception as e:
|
|
self._json(500, {"ok": False, "error": str(e)})
|
|
|
|
def _json(self, code, obj):
|
|
body = json.dumps(obj).encode()
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def launch(domain, cookies, tok, proxy=""):
|
|
from selenium import webdriver
|
|
from selenium.webdriver.chrome.options import Options
|
|
opts = Options()
|
|
profile = os.path.join(SESSION_DIR, re.sub(r'[^a-zA-Z0-9]', '_', domain))
|
|
os.makedirs(profile, exist_ok=True)
|
|
opts.add_argument(r"--user-data-dir=" + profile)
|
|
opts.add_argument("--no-first-run")
|
|
opts.add_argument("--no-default-browser-check")
|
|
opts.add_argument("--start-maximized")
|
|
opts.add_argument("--window-size=1440,900")
|
|
opts.add_argument("--disable-gpu")
|
|
if proxy:
|
|
opts.add_argument("--proxy-server=" + proxy)
|
|
from selenium.webdriver.chrome.service import Service
|
|
svc = Service(executable_path=os.path.join(SESSION_DIR, "..", "chromedriver.exe"))
|
|
driver = webdriver.Chrome(service=svc, options=opts)
|
|
try:
|
|
SCREENSHOTS[tok] = _png(driver, "starting")
|
|
host = domain.lstrip(".")
|
|
try:
|
|
driver.get(f"https://{host}/favicon.ico")
|
|
except Exception:
|
|
try: driver.get(f"https://{host}/")
|
|
except Exception: pass
|
|
time.sleep(1.5)
|
|
added = 0
|
|
for ck in cookies:
|
|
try:
|
|
c = {"name": ck.get("name"), "value": ck.get("value", ""),
|
|
"domain": ck.get("domain", ""), "path": ck.get("path", "/") or "/",
|
|
"secure": bool(ck.get("secure"))}
|
|
if not c["name"]:
|
|
continue
|
|
exp = ck.get("expiry") or 0
|
|
try: exp = int(float(exp))
|
|
except Exception: exp = 0
|
|
if exp > 0: c["expiry"] = exp
|
|
try:
|
|
driver.add_cookie(c); added += 1
|
|
except Exception:
|
|
c2 = dict(c); c2.pop("expiry", None)
|
|
try: driver.add_cookie(c2); added += 1
|
|
except Exception: pass
|
|
except Exception:
|
|
pass
|
|
try:
|
|
driver.get(f"https://{host}/")
|
|
time.sleep(4)
|
|
SCREENSHOTS[tok] = _png(driver, "done")
|
|
except Exception:
|
|
# screenshot whatever state we're in — even a failed load shows something
|
|
try:
|
|
driver.get(f"https://{host}/")
|
|
time.sleep(3)
|
|
SCREENSHOTS[tok] = _png(driver, "retry")
|
|
except Exception:
|
|
SCREENSHOTS[tok] = _error_png(f"page load failed: {host}")
|
|
with open(os.path.join(SESSION_DIR, "_launch.log"), "a", encoding="utf-8") as f:
|
|
f.write(f"{time.ctime()} {domain}: added {added}/{len(cookies)} cookies via {proxy or 'direct'}\n")
|
|
except Exception as e:
|
|
SCREENSHOTS.setdefault(tok, _error_png(f"launch error: {e}"))
|
|
with open(os.path.join(SESSION_DIR, "_launch.log"), "a", encoding="utf-8") as f:
|
|
f.write(f"{time.ctime()} {domain}: LAUNCH ERROR {e}\n")
|
|
finally:
|
|
try: driver.quit()
|
|
except Exception: pass
|
|
|
|
def check_login(domain, cookies, proxy=""):
|
|
"""Load the site with these cookies headlessly, detect login state. Returns dict."""
|
|
from selenium import webdriver
|
|
from selenium.webdriver.chrome.options import Options
|
|
opts = Options()
|
|
opts.add_argument("--headless=new")
|
|
opts.add_argument("--no-first-run")
|
|
opts.add_argument("--no-default-browser-check")
|
|
opts.add_argument("--window-size=1280,800")
|
|
opts.add_argument("--disable-gpu")
|
|
import tempfile
|
|
prof = tempfile.mkdtemp(prefix="chk_", dir=SESSION_DIR)
|
|
opts.add_argument(r"--user-data-dir=" + prof)
|
|
if proxy:
|
|
opts.add_argument("--proxy-server=" + proxy)
|
|
from selenium.webdriver.chrome.service import Service
|
|
svc = Service(executable_path=os.path.join(SESSION_DIR, "..", "chromedriver.exe"))
|
|
driver = None
|
|
try:
|
|
driver = webdriver.Chrome(service=svc, options=opts)
|
|
host = domain.lstrip(".")
|
|
try:
|
|
driver.get(f"https://{host}/")
|
|
except Exception:
|
|
return {"status": "dead", "detail": "site unreachable"}
|
|
time.sleep(1.0)
|
|
added = 0
|
|
for ck in cookies:
|
|
try:
|
|
c = {"name": ck.get("name"), "value": ck.get("value", ""),
|
|
"domain": ck.get("domain", ""), "path": ck.get("path", "/") or "/",
|
|
"secure": bool(ck.get("secure"))}
|
|
if not c["name"]:
|
|
continue
|
|
exp = ck.get("expiry") or 0
|
|
try: exp = int(float(exp))
|
|
except Exception: exp = 0
|
|
if exp > 0: c["expiry"] = exp
|
|
try:
|
|
driver.add_cookie(c); added += 1
|
|
except Exception:
|
|
c2 = dict(c); c2.pop("expiry", None)
|
|
try: driver.add_cookie(c2); added += 1
|
|
except Exception: pass
|
|
except Exception:
|
|
pass
|
|
try:
|
|
driver.get(f"https://{host}/")
|
|
except Exception:
|
|
return {"status": "dead", "detail": "reload failed"}
|
|
time.sleep(2.5)
|
|
body = ""
|
|
try:
|
|
body = driver.page_source[:200000].lower()
|
|
except Exception:
|
|
pass
|
|
# cookie-jar proof: did OUR cookies actually land in the browser jar?
|
|
jar_names = set()
|
|
try:
|
|
for ck in driver.get_cookies():
|
|
jar_names.add(ck.get("name", ""))
|
|
except Exception:
|
|
pass
|
|
sent_names = {ck.get("name") for ck in cookies if ck.get("name")}
|
|
kept = len(sent_names & jar_names)
|
|
# login detection heuristics
|
|
logged_in_markers = ["sign out", "log out", "logout", "my account", "account settings",
|
|
"signed in as", "avatar-btn", "img-avatar", "profile-menu",
|
|
"creator studio", "youtube studio", "switch account",
|
|
"ggpht.com", "googleusercontent", "authuser=", "/u/0/",
|
|
"authenticated", "welcome back", "account-icon", "avatar-image",
|
|
"data-avatar", "member-menu", "user-menu", "my profile"]
|
|
logged_out_markers = ["sign in", "log in", "create account", "register now",
|
|
"signup", "sign-up"]
|
|
li = sum(1 for m in logged_in_markers if m in body)
|
|
lo = sum(1 for m in logged_out_markers if m in body)
|
|
# "login" alone is ambiguous (appears in code), require 2+ strong logout markers
|
|
if "sign in" in body and "log in" in body:
|
|
lo += 1
|
|
title = ""
|
|
try: title = driver.title[:120]
|
|
except Exception: pass
|
|
if li >= 2 and li > lo:
|
|
status = "ACTIVE"
|
|
elif li > 0 and lo == 0:
|
|
status = "ACTIVE?"
|
|
elif lo >= 2:
|
|
status = "dead"
|
|
else:
|
|
status = "unclear"
|
|
# For JS-heavy sites (YouTube etc.) the DOM never shows markers in raw HTML.
|
|
# Decisive signal = session cookies were RETAINED after reload (sites drop dead
|
|
# session cookies on reload when the server rejects them).
|
|
if status == "unclear" and sent_names:
|
|
if kept >= max(2, len(sent_names) // 2):
|
|
status = "ACTIVE?"
|
|
detail_jar = f"jar kept {kept}/{len(sent_names)} session cookies"
|
|
elif kept == 0:
|
|
status = "dead"
|
|
detail_jar = f"jar kept 0/{len(sent_names)} — server dropped session"
|
|
else:
|
|
status = "unclear"
|
|
detail_jar = f"jar kept {kept}/{len(sent_names)}"
|
|
else:
|
|
detail_jar = f"jar {kept}/{len(sent_names)}" if sent_names else ""
|
|
return {"status": status, "detail": (f"{detail_jar} title={title}" if detail_jar else f"title={title}"),
|
|
"added": added,
|
|
"shot": base64.b64encode(_png(driver, "check")).decode() if _png(driver, "check") else ""}
|
|
except Exception as e:
|
|
return {"status": "error", "detail": str(e)[:200]}
|
|
finally:
|
|
if driver:
|
|
try: driver.quit()
|
|
except Exception: pass
|
|
try:
|
|
import shutil
|
|
shutil.rmtree(prof, ignore_errors=True)
|
|
except Exception:
|
|
pass
|
|
|
|
def _png(driver, stage):
|
|
try:
|
|
return driver.get_screenshot_as_png()
|
|
except Exception:
|
|
return b""
|
|
|
|
_ERR_PNG = base64.b64decode(
|
|
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg==")
|
|
|
|
def _error_png(msg):
|
|
with open(os.path.join(SESSION_DIR, "_errors.log"), "a", encoding="utf-8") as f:
|
|
f.write(f"{time.ctime()} {msg}\n")
|
|
return _ERR_PNG
|
|
|
|
SCREENSHOTS = {}
|
|
|
|
if __name__ == "__main__":
|
|
HTTPServer(("127.0.0.1", 5067), Handler).serve_forever()
|