diff --git a/app.py b/app.py index 3f283a6..701768a 100644 --- a/app.py +++ b/app.py @@ -1,22 +1,25 @@ #!/usr/bin/env python3 -# COOKIE VAULT v2 — Netscape cookies.txt importer + one-click LOGIN launcher -# Host: Commando VM601, C:\cookievault\app.py, port 5066 -import os, re, json, sqlite3, time, datetime, threading, subprocess, tempfile +# COOKIE VAULT v5 — clean UI, session-1 chrome launch bridge, LLM fallback parse +import os, re, json, sqlite3, time, datetime, threading, subprocess, urllib.request from flask import Flask, request, render_template_string, redirect, url_for, jsonify, Response APP_DIR = r"C:\cookievault" DB_PATH = os.path.join(APP_DIR, "cookies.db") WATCH_DIR = os.path.join(APP_DIR, "incoming") SESSION_DIR = os.path.join(APP_DIR, "sessions") -WORK_DRIVE = r"D:\\" # the "Work" removable drive -SCAN_ROOTS = [WATCH_DIR, WORK_DRIVE] +WORK_DRIVE = r"D:\\" +BRIDGE = "http://127.0.0.1:5067/launch" +OLLAMA_HOSTS = ["10.30.20.222", "10.30.20.29"] +LLM_MODEL = "ornith-1.5:9b-64k" os.makedirs(WATCH_DIR, exist_ok=True) os.makedirs(SESSION_DIR, exist_ok=True) app = Flask(__name__) +DOMAIN_RE = re.compile(r"^\.?([a-z0-9]([a-z0-9-]*[a-z0-9])?\.)+[a-z]{2,}$", re.I) + def db(): - c = sqlite3.connect(DB_PATH) + c = sqlite3.connect(DB_PATH, timeout=30) c.execute("""CREATE TABLE IF NOT EXISTS cookies( id INTEGER PRIMARY KEY AUTOINCREMENT, domain TEXT, flag TEXT, path TEXT, secure TEXT, @@ -27,10 +30,15 @@ def db(): return c def utc(ts): - try: return datetime.datetime.utcfromtimestamp(int(ts)).strftime("%Y-%m-%d") + try: return datetime.datetime.utcfromtimestamp(int(float(ts))).strftime("%Y-%m-%d") except Exception: return "session" -NS_RE = re.compile(r"^(#\S+)?\s*(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(.*)$") +def _safe_expiry(val): + try: + iv = int(float(val)) + return iv if iv > 0 else 0 + except (TypeError, ValueError): + return 0 def parse_netscape(text): out = [] @@ -51,28 +59,60 @@ def parse_netscape(text): out.append((domain, flag, path, secure, expiry, name, value)) return out -def _safe_expiry(val): - try: - iv = int(float(val)) - return iv if iv > 0 else 0 - except (TypeError, ValueError): - return 0 +def sniff_netscape(text): + """Deterministic: does the first data line look like 7-col cookie format?""" + seen = 0 + for line in text.splitlines(): + line = line.strip() + if not line or line.startswith("#"): + continue + parts = line.split("\t") if "\t" in line else line.split() + if len(parts) >= 7 and parts[4].isdigit(): + return True + seen += 1 + if seen > 5: + return False + return False -def import_file(path): +def llm_classify(text): + """Fallback: ask local ornith whether the file holds cookie data. Returns True/False/None(fail).""" + prompt = ("You are a file classifier. Does this text contain browser cookie data in Netscape " + "format (7 tab-separated fields: domain, TRUE/FALSE, path, TRUE/FALSE, numeric expiry, name, value)? " + "Answer exactly one word: YES or NO.\n\n" + text[:2000]) + for host in OLLAMA_HOSTS: + try: + req = urllib.request.Request(f"http://{host}:11434/api/generate", + data=json.dumps({"model": LLM_MODEL, "prompt": prompt, "stream": False, "think": False}).encode(), + headers={"Content-Type": "application/json"}) + with urllib.request.urlopen(req, timeout=45) as r: + ans = json.load(r)["response"].strip().upper() + if "YES" in ans[:6]: return True + if "NO" in ans[:6]: return False + if ans and "?" not in ans[:6]: return False + except Exception: + continue + return None + +def import_file(path, use_llm_fallback=True): fname = os.path.basename(path) with open(path, "r", encoding="utf-8", errors="replace") as f: text = f.read() + if not sniff_netscape(text): + if not (use_llm_fallback and llm_classify(text)): + return 0, 0, None # not a cookie file cookies = parse_netscape(text) - # guard: every cookie must have sane types (name+value non-empty, expiry numeric-ish) clean = [] for ck in cookies: if not ck[5] or not ck[6]: continue - if re.match(r"^[0-9]+$", ck[4].strip() or "0") is None: - # header junk line that survived parsing (e.g. split words) — skip + if not re.match(r"^[0-9]+$", (ck[4] or "0").strip()): + continue + if not DOMAIN_RE.match(ck[0] or ""): continue clean.append(ck) cookies = clean + if not cookies: + return 0, 0, None c = db() before = c.execute("SELECT COUNT(*) FROM cookies").fetchone()[0] for ck in cookies: @@ -80,63 +120,89 @@ def import_file(path): (*ck, fname, int(time.time()))) c.commit() after = c.execute("SELECT COUNT(*) FROM cookies").fetchone()[0] - total = after c.close() - return len(cookies), after - before, total + return len(cookies), after - before, after -PAGE = r"""COOKIE VAULT -

🍪 COOKIE VAULT

{{stats[0]}} cookies{{stats[1]}} domains{{stats[2]}} files
-
-
-

Import Netscape cookies.txt

-
-
Drop cookies.txt files here (as many as you want)  or  - -

Or paste raw cookie text:

- -
-

also watches C:\cookievault\incoming — dump files there and hit rescan

+details summary{cursor:pointer;color:#7aa2f7;font-weight:600;padding:8px 0} +""" + +HOME_PAGE = STYLE + r"""COOKIE VAULT +
+
🍪 COOKIE VAULT
+
{{stats[0]}} cookies{{stats[1]}} domains{{stats[2]}} files
+
+
-
-

Domains — one-click login

-
- ALL - EXPORT JSON - EXPORT NETSCAPE
- - {% for d in domains %} - - - - {% endfor %}
domaincookiesfresh untillogin
{{d[0]}}{{d[1]}}{{'has session cookies' if d[2]==0 else utc(d[2])}}LOGIN →
-

{{domains|length}} domains shown

+ 📁 BROWSE WORK DRIVE + 📥 IMPORT FILES + 💾 EXPORT JSON + 💾 EXPORT TXT + +
+{% if msg %}
{{msg|safe}}
{% endif %} + + +{% for d in domains %} + + + + + + +{% endfor %} +
domaincookiesfresh until
{{d[0]}}{{d[1]}}{{utc(d[2]) if d[2] else 'session'}}LOGIN →
+
+{% if page>1 %}« prev{% endif %} +{{page}} +{% if has_more %}next »{% endif %}
+
""" + +IMPORT_PAGE = STYLE + r"""Import — COOKIE VAULT +
🍪 COOKIE VAULT
← back
+
+
Drop cookie files below, browse the Work drive, or paste raw text.
+
+
Drag cookie files here — any filename, any format — or + + +

+
+

+ +

+
""" -BROWSE_PAGE = r"""Pick a folder — COOKIE VAULT -

📁 Pick a folder on WORK (D:\)

-
D:\ {% if rel != '.' %}» {{rel}}{% endif %}
+BROWSE_PAGE = STYLE + r"""Pick folder — COOKIE VAULT +
📁 WORK DRIVE — D:\{{' » ' + rel if rel != '.'}}
+← back
-{% if parent is not none %}
← UP
{% endif %} -
    -{% for e in entries %}
  • 📁 {{e}}/
    • {% endfor %} +{% if parent is not none %}
    ← UP
    {% endif %} +
      +{% for e in entries %}
    • + 📁 {{e}}/
      • {% endfor %} {% if not entries %}
    • no subfolders here
      • {% endif %}
    -
    +
    -
    - Cancel +
    -

    Scans this folder + all subfolders. Any .txt that contains Netscape cookies (any filename) gets imported.

    +

    Scans this folder + subfolders. Any file with cookie data gets imported — any filename, any format (LLM-assisted detection).

""" +PAGE_SIZE = 40 + @app.route("/", methods=["GET"]) def index(): q = request.args.get("q", "").strip() + try: page = max(1, int(request.args.get("page", 1))) + except ValueError: page = 1 + like = f"%{q}%" if q else "%" c = db() if q: - like = f"%{q}%" - domains = c.execute("""SELECT domain, COUNT(*), MAX(CASE WHEN expiry GLOB '[0-9]*' THEN CAST(expiry AS INTEGER) ELSE 0 END) FROM cookies - WHERE domain LIKE ? GROUP BY domain ORDER BY domain LIMIT 500""", (like,)).fetchall() + total = c.execute("SELECT COUNT(DISTINCT domain) FROM cookies WHERE domain LIKE ?", (like,)).fetchone()[0] + domains = c.execute("""SELECT domain, COUNT(*), MAX(CASE WHEN expiry GLOB '[0-9]*' THEN CAST(expiry AS INTEGER) ELSE 0 END) + FROM cookies WHERE domain LIKE ? GROUP BY domain + HAVING COUNT(*) >= 2 AND domain GLOB '*.*.*' OR (domain LIKE ? AND COUNT(*) >= 1) + ORDER BY COUNT(*) DESC, domain LIMIT ? OFFSET ?""", + (like, like, PAGE_SIZE, (page-1)*PAGE_SIZE)).fetchall() else: - domains = c.execute("""SELECT domain, COUNT(*), MAX(CASE WHEN expiry GLOB '[0-9]*' THEN CAST(expiry AS INTEGER) ELSE 0 END) FROM cookies - GROUP BY domain ORDER BY domain LIMIT 500""").fetchall() + total = c.execute("SELECT COUNT(DISTINCT domain) FROM cookies").fetchone()[0] + domains = c.execute("""SELECT domain, COUNT(*), MAX(CASE WHEN expiry GLOB '[0-9]*' THEN CAST(expiry AS INTEGER) ELSE 0 END) + FROM cookies GROUP BY domain + HAVING COUNT(*) >= 3 AND domain GLOB '*.*.*' + ORDER BY COUNT(*) DESC, domain LIMIT ? OFFSET ?""", + (PAGE_SIZE, (page-1)*PAGE_SIZE)).fetchall() stats = (c.execute("SELECT COUNT(*) FROM cookies").fetchone()[0], c.execute("SELECT COUNT(DISTINCT domain) FROM cookies").fetchone()[0], c.execute("SELECT COUNT(DISTINCT source_file) FROM cookies").fetchone()[0]) c.close() - return render_template_string(PAGE, domains=domains, stats=stats, q=q, msg=request.args.get("msg"), utc=utc) + return render_template_string(HOME_PAGE, domains=domains, stats=stats, q=q, page=page, + has_more=(page*PAGE_SIZE) < total, msg=request.args.get("msg"), utc=utc) + +@app.route("/import_page") +def import_page(): + return render_template_string(IMPORT_PAGE, msg=request.args.get("msg")) @app.route("/import", methods=["POST"]) def do_import(): msg, errs = [], [] - files = request.files.getlist("files") - for f in files: + for f in request.files.getlist("files"): try: fname = os.path.basename(f.filename or "").strip() - # sanitize: keep alnum, dash, underscore, dot fname = re.sub(r"[^A-Za-z0-9._\- ]", "_", fname) if not fname or fname == ".": - # no filename — generate one from content hash data = f.read() - if not data.strip(): - continue - fname = "unnamed_%d.txt" % int(time.time() * 1000 % 1000000000) + if not data.strip(): continue + fname = "unnamed_%d.txt" % (int(time.time()*1000) % 10**9) path = os.path.join(WATCH_DIR, fname) with open(path, "wb") as fh: fh.write(data) else: path = os.path.join(WATCH_DIR, fname) f.save(path) n, new, total = import_file(path) - msg.append(f"{fname}: {n} parsed, {new} new") + msg.append(f"{fname}: {new} new" if total is not None else f"{fname}: no cookie data found") except Exception as e: - errs.append(f"{getattr(f, 'filename', '?')}: {e}") + errs.append(f"{getattr(f,'filename','?')}: {e}") paste = request.form.get("paste", "").strip() if paste: - try: - tmp = os.path.join(WATCH_DIR, "_pasted_%d.txt" % int(time.time())) - with open(tmp, "w", encoding="utf-8") as fh: fh.write(paste) - n, new, total = import_file(tmp) - msg.append(f"pasted: {n} parsed, {new} new") - except Exception as e: - errs.append(f"paste: {e}") - if errs: - msg += ["%s" % e for e in errs] - if not msg: - msg = ["nothing imported — pick a cookies.txt file or paste cookie text first"] - return redirect(url_for("index", msg=" · ".join(msg))) + tmp = os.path.join(WATCH_DIR, "_pasted_%d.txt" % int(time.time())) + with open(tmp, "w", encoding="utf-8") as fh: fh.write(paste) + n, new, total = import_file(tmp) + msg.append(f"pasted: {new} new" if total is not None else "pasted: no cookie data detected") + msg += ["%s" % e for e in errs] + return redirect(url_for("index", msg=" · ".join(msg) or "nothing imported")) @app.route("/scan", methods=["POST"]) def scan(): - return _scan_dirs(SCAN_ROOTS) + msg = [] + for fn in os.listdir(WATCH_DIR): + if fn.lower().endswith((".txt", ".json")) and not fn.startswith("_pasted"): + n, new, total = import_file(os.path.join(WATCH_DIR, fn)) + msg.append(f"{fn}: {new} new") + return redirect(url_for("index", msg=" · ".join(msg) or "nothing new")) -@app.route("/scan_work", methods=["POST"]) -def scan_work(): - """Deep-scan the Work drive: every .txt file that LOOKS like Netscape cookie format.""" - return _smart_scan(WORK_DRIVE) - -@app.route("/browse", methods=["GET"]) +@app.route("/browse") def browse(): - """List subfolders of a path on the Work drive so the user can pick one.""" sub = request.args.get("path", "").strip() base = os.path.abspath(WORK_DRIVE) target = os.path.abspath(os.path.join(base, sub.lstrip("/\\"))) if sub else base if not target.startswith(base) or not os.path.isdir(target): - return redirect(url_for("index", msg="invalid folder")) + return redirect(url_for("index", msg="invalid folder")) entries = [] - try: - for name in sorted(os.listdir(target)): - full = os.path.join(target, name) - if os.path.isdir(full) and name not in ("$RECYCLE.BIN", "System Volume Information"): - entries.append(name) - except Exception as e: - return redirect(url_for("index", msg=f"{e}")) + for name in sorted(os.listdir(target)): + if os.path.isdir(os.path.join(target, name)) and name not in ("$RECYCLE.BIN", "System Volume Information"): + entries.append(name) rel = os.path.relpath(target, base) parent = os.path.dirname(rel) if rel != "." else None return render_template_string(BROWSE_PAGE, entries=entries, rel=rel, parent=parent) @@ -269,133 +322,115 @@ def scan_folder(): base = os.path.abspath(WORK_DRIVE) target = os.path.abspath(os.path.join(base, sub.lstrip("/\\"))) if sub else base if not target.startswith(base) or not os.path.isdir(target): - return redirect(url_for("index", msg="invalid folder")) + return redirect(url_for("index", msg="invalid folder")) return _smart_scan(target) +@app.route("/scan_work", methods=["POST"]) +def scan_work(): + if not os.path.isdir(WORK_DRIVE): + return redirect(url_for("index", msg="Work drive (D:\\) not plugged in")) + return _smart_scan(WORK_DRIVE) + def _smart_scan(root): - """Smart scan: walk root, sniff every .txt for Netscape format regardless of filename.""" found = [] for rootpath, dirs, files in os.walk(root): dirs[:] = [d for d in dirs if d not in ("$RECYCLE.BIN", "System Volume Information", "node_modules")] for fn in files: - if fn.lower().endswith(".txt"): + if fn.lower().endswith((".txt", ".json", ".log")): found.append(os.path.join(rootpath, fn)) if not found: - return redirect(url_for("index", msg="no .txt files found in that folder")) + return redirect(url_for("index", msg="no text files found in that folder")) msg, errs = [], [] - new_total, imported_files = 0, 0 + new_total, imported_files, llm_used = 0, 0, 0 for p in found: try: - looks_like = False with open(p, "r", encoding="utf-8", errors="replace") as f: - for i, line in enumerate(f): - line = line.strip() - if not line or line.startswith("#"): - continue - parts = line.split("\t") if "\t" in line else line.split() - if len(parts) >= 7 and parts[4].isdigit(): - looks_like = True - if looks_like or i > 30: - break - if not looks_like: - continue + text = f.read(65536) + if not sniff_netscape(text): + # deterministic miss -> try LLM only for small-ish files (cost control) + if os.path.getsize(p) > 200_000: + continue + verdict = llm_classify(text) + if verdict is None: + continue + llm_used += 1 + if not verdict: + continue + with open(p, "r", encoding="utf-8", errors="replace") as f: + pass # full read happens in import_file n, new, total = import_file(p) + if total is None: + continue imported_files += 1 new_total += new - try: - shown = os.path.relpath(p, root) - except ValueError: - shown = p + try: shown = os.path.relpath(p, root) + except ValueError: shown = p msg.append(f"{shown}: {new} new") except Exception as e: errs.append(f"{p}: {e}") if not imported_files and not errs: - return redirect(url_for("index", msg=f"scanned {len(found)} .txt files — none were Netscape cookie format")) + return redirect(url_for("index", msg=f"scanned {len(found)} files — no cookie data found (LLM checked {llm_used} ambiguous ones)")) msg += ["%s" % e for e in errs] - return redirect(url_for("index", msg=f"Scan of {root}: {imported_files} cookie files of {len(found)} txt, {new_total} new cookies · " + " · ".join(msg[:25]))) + return redirect(url_for("index", msg=f"{imported_files} cookie files ({new_total} new cookies, LLM-assisted on {llm_used}) · " + " · ".join(msg[:20]))) -def _scan_dirs(dirs): - msg, errs = [], [] - for d in dirs: - if not os.path.isdir(d): - continue - for fn in os.listdir(d): - if fn.lower().endswith((".txt", ".json")) and not fn.startswith("_pasted"): - try: - n, new, total = import_file(os.path.join(d, fn)) - msg.append(f"{fn}: {new} new") - except Exception as e: - errs.append(f"{fn}: {e}") - msg += ["%s" % e for e in errs] - return redirect(url_for("index", msg=" · ".join(msg) or "nothing new found")) - -def _launch_login(domain, cookies): - """Runs in background thread: selenium Chrome with cookies injected.""" - 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") - driver = webdriver.Chrome(options=opts) # selenium-manager auto-fetches driver - try: - # land on the site's origin first so cookies can be set for it - host = domain.lstrip(".") - driver.get(f"https://{host}/favicon.ico") - except Exception: - try: driver.get(f"https://{host}/") - except Exception: pass - time.sleep(1) - for ck in cookies: - c = {"name": ck[1], "value": ck[2]} - c["domain"] = ck[0] - c["path"] = ck[3] or "/" - exp = _safe_expiry(ck[4]) - if exp > 0: - c["expiry"] = exp - c["secure"] = str(ck[5]).upper() == "TRUE" - try: - driver.add_cookie(c) - except Exception: - try: - c2 = dict(c); c2.pop("expiry", None) - driver.add_cookie(c2) - except Exception: - pass - try: - driver.get(f"https://{host}/") - except Exception: - pass - # keep the process ref alive — selenium closes browser if driver is GC'd - globals().setdefault("_drivers", []).append(driver) +# ---------------- LOGIN (session-1 bridge) ---------------- @app.route("/open") def open_login(): domain = request.args.get("domain", "").strip() if not domain: return redirect(url_for("index")) - like = f"%{domain.strip('.')}%" + stem = domain.strip(".") + like = stem + "%" c = db() rows = c.execute("""SELECT domain, name, value, path, expiry, secure FROM cookies - WHERE domain LIKE ? OR domain LIKE ?""", - (domain.strip(".") + "%", "%" + domain.strip("."))).fetchall() + WHERE domain LIKE ? OR domain LIKE ? ORDER BY (domain = ?) DESC LIMIT 4000""", + (like, "%" + stem, domain)).fetchall() c.close() if not rows: - return redirect(url_for("index", msg=f"no cookies stored for {domain}")) - threading.Thread(target=_launch_login, args=(domain, rows), daemon=True).start() - time.sleep(0.5) - return redirect(url_for("index", msg=f"launching Chrome for {domain} with {len(rows)} cookies...")) + return redirect(url_for("index", msg=f"no cookies for {domain}")) + payload = {"domain": domain, "cookies": [ + {"domain": r[0], "name": r[1], "value": r[2], "path": r[3], + "expiry": _safe_expiry(r[4]), "secure": str(r[5]).upper() == "TRUE"} for r in rows]} + try: + req = urllib.request.Request(BRIDGE, data=json.dumps(payload).encode(), + headers={"Content-Type": "application/json"}) + with urllib.request.urlopen(req, timeout=15) as r: + res = json.load(r) + if res.get("ok"): + return render_template_string(LAUNCH_PAGE, domain=domain, n=len(rows), + token=res.get("token", ""), url=f"https://{stem}/") + return redirect(url_for("index", msg=f"bridge error: {res.get('error')}")) + except Exception as e: + return redirect(url_for("index", msg=f"launcher bridge not reachable ({e})")) + +LAUNCH_PAGE = STYLE + r"""Launching… + +
🎯 logging in to {{domain}}
+
+
Launching Chrome with {{n}} cookies… this page auto-refreshes; the screenshot appears below when the session is ready.
+{% if token %}{% endif %} +
← back to vault
+
""" + +@app.route("/shot/") +def shot(token): + token = re.sub(r"[^a-z0-9_]", "", token) + try: + with urllib.request.urlopen(f"http://127.0.0.1:5067/shot/{token}", timeout=10) as r: + img = r.read() + return Response(img, mimetype="image/png") + except Exception: + return Response(b"", mimetype="image/png", status=503) + +# ---------------- EXPORT ---------------- @app.route("/export") def export(): q = request.args.get("q", "").strip() c = db() if q: - like = f"%{q}%" - rows = c.execute("SELECT domain,name,value,path,expiry,secure FROM cookies WHERE domain LIKE ? OR name LIKE ?", (like, like)).fetchall() + rows = c.execute("SELECT domain,name,value,path,expiry,secure FROM cookies WHERE domain LIKE ? OR name LIKE ?", (f"%{q}%", f"%{q}%")).fetchall() else: rows = c.execute("SELECT domain,name,value,path,expiry,secure FROM cookies").fetchall() c.close() @@ -410,13 +445,12 @@ def export_ns(): q = request.args.get("q", "").strip() c = db() if q: - like = f"%{q}%" - rows = c.execute("SELECT domain,flag,path,secure,expiry,name,value FROM cookies WHERE domain LIKE ? OR name LIKE ?", (like, like)).fetchall() + rows = c.execute("SELECT domain,flag,path,secure,expiry,name,value FROM cookies WHERE domain LIKE ?", (f"%{q}%",)).fetchall() else: rows = c.execute("SELECT domain,flag,path,secure,expiry,name,value FROM cookies").fetchall() c.close() lines = ["# Netscape HTTP Cookie File", "# Exported by COOKIE VAULT"] - lines += ["\t".join(str(x) for x in r) for r in rows if _safe_expiry(r[4]) >= 0] + lines += ["\t".join(str(x) for x in r) for r in rows] return Response("\n".join(lines) + "\n", mimetype="text/plain", headers={"Content-Disposition": "attachment; filename=cookies.txt"}) diff --git a/launcher_bridge.py b/launcher_bridge.py new file mode 100644 index 0000000..6df3252 --- /dev/null +++ b/launcher_bridge.py @@ -0,0 +1,113 @@ +#!/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 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 != "/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", []) + 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), 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): + 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") + driver = webdriver.Chrome(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: + SCREENSHOTS[tok] = _png(driver, "cookies-set") + 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\n") + finally: + try: driver.quit() + except Exception: pass + +def _png(driver, stage): + try: + return driver.get_screenshot_as_png() + except Exception: + return b"" + +SCREENSHOTS = {} + +if __name__ == "__main__": + HTTPServer(("127.0.0.1", 5067), Handler).serve_forever() diff --git a/start_bridge.bat b/start_bridge.bat new file mode 100644 index 0000000..9f0a0a3 --- /dev/null +++ b/start_bridge.bat @@ -0,0 +1,3 @@ +@echo off +cd /d C:\cookievault +python launcher_bridge.py >> C:\cookievault\bridge.log 2>&1