#!/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
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]
os.makedirs(WATCH_DIR, exist_ok=True)
os.makedirs(SESSION_DIR, exist_ok=True)
app = Flask(__name__)
def db():
c = sqlite3.connect(DB_PATH)
c.execute("""CREATE TABLE IF NOT EXISTS cookies(
id INTEGER PRIMARY KEY AUTOINCREMENT,
domain TEXT, flag TEXT, path TEXT, secure TEXT,
expiry INTEGER, name TEXT, value TEXT,
source_file TEXT, imported_at INTEGER,
UNIQUE(domain, path, name, source_file))""")
c.execute("CREATE INDEX IF NOT EXISTS ix_dom ON cookies(domain)")
return c
def utc(ts):
try: return datetime.datetime.utcfromtimestamp(int(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 parse_netscape(text):
out = []
for line in text.splitlines():
line = line.strip()
if not line or line.startswith("# ") or line == "#HttpOnly_":
continue
if line.startswith("#HttpOnly_"):
line = line[len("#HttpOnly_"):]
elif line.startswith("#"):
continue
parts = line.split("\t") if "\t" in line else line.split()
if len(parts) < 7:
continue
domain, flag, path, secure, expiry, name, value = parts[:7]
if len(parts) > 7:
value = "\t".join(parts[6:])
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 import_file(path):
fname = os.path.basename(path)
with open(path, "r", encoding="utf-8", errors="replace") as f:
text = f.read()
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
continue
clean.append(ck)
cookies = clean
c = db()
before = c.execute("SELECT COUNT(*) FROM cookies").fetchone()[0]
for ck in cookies:
c.execute("INSERT OR IGNORE INTO cookies(domain,flag,path,secure,expiry,name,value,source_file,imported_at) VALUES(?,?,?,?,?,?,?,?,?)",
(*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
PAGE = r"""
Scans this folder + all subfolders. Any .txt that contains Netscape cookies (any filename) gets imported.
"""
@app.route("/", methods=["GET"])
def index():
q = request.args.get("q", "").strip()
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()
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()
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)
@app.route("/import", methods=["POST"])
def do_import():
msg, errs = [], []
files = request.files.getlist("files")
for f in 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)
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")
except Exception as 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)))
@app.route("/scan", methods=["POST"])
def scan():
return _scan_dirs(SCAN_ROOTS)
@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"])
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"))
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}"))
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)
@app.route("/scan_folder", methods=["POST"])
def scan_folder():
sub = request.form.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 _smart_scan(target)
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"):
found.append(os.path.join(rootpath, fn))
if not found:
return redirect(url_for("index", msg="no .txt files found in that folder"))
msg, errs = [], []
new_total, imported_files = 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
n, new, total = import_file(p)
imported_files += 1
new_total += new
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"))
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])))
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)
@app.route("/open")
def open_login():
domain = request.args.get("domain", "").strip()
if not domain:
return redirect(url_for("index"))
like = f"%{domain.strip('.')}%"
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()
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..."))
@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()
else:
rows = c.execute("SELECT domain,name,value,path,expiry,secure FROM cookies").fetchall()
c.close()
jar = [{"domain": r[0], "name": r[1], "value": r[2], "path": r[3],
"expirationDate": _safe_expiry(r[4]) or None,
"secure": str(r[5]).upper() == "TRUE", "httpOnly": False} for r in rows]
return Response(json.dumps(jar, indent=2), mimetype="application/json",
headers={"Content-Disposition": "attachment; filename=cookies.json"})
@app.route("/export_netscape")
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()
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]
return Response("\n".join(lines) + "\n", mimetype="text/plain",
headers={"Content-Disposition": "attachment; filename=cookies.txt"})
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5066, debug=False)