diff --git a/app.py b/app.py index 2a085cb..3f283a6 100644 --- a/app.py +++ b/app.py @@ -8,6 +8,8 @@ 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) @@ -49,11 +51,28 @@ 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 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: @@ -98,7 +117,10 @@ td.val{color:#9ece6a;max-width:240px;overflow:hidden;text-overflow:ellipsis;whit

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

-
+
+ 📁 PICK FOLDER ON WORK DRIVE +
+ smart scan: any .txt containing Netscape cookies gets imported, whatever its name {% if msg %}

{{msg|safe}}

{% endif %}
@@ -124,16 +146,48 @@ drop.addEventListener('drop',ev=>{m.files=ev.dataTransfer.files;document.getElem m.addEventListener('change',()=>document.getElementById('f').submit()); """ +BROWSE_PAGE = r"""Pick a folder — COOKIE VAULT +

📁 Pick a folder on WORK (D:\)

+
D:\ {% if rel != '.' %}» {{rel}}{% endif %}
+
+{% if parent is not none %}
← UP
{% endif %} + +
+
+
+ Cancel +
+

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(CAST(expiry AS INTEGER)) FROM cookies + 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(CAST(expiry AS INTEGER)) FROM cookies + 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], @@ -182,16 +236,98 @@ def do_import(): @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 = [], [] - for fn in os.listdir(WATCH_DIR): - if fn.lower().endswith((".txt", ".json")): + 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: - n, new, total = import_file(os.path.join(WATCH_DIR, fn)) - msg.append(f"{fn}: {new} new") - except Exception as e: - errs.append(f"{fn}: {e}") + 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=" · ".join(msg) or "nothing new in incoming")) + 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.""" @@ -215,18 +351,11 @@ def _launch_login(domain, cookies): time.sleep(1) for ck in cookies: c = {"name": ck[1], "value": ck[2]} - dom = ck[0] - if not dom.startswith("."): - c["domain"] = dom - else: - c["domain"] = dom + c["domain"] = ck[0] c["path"] = ck[3] or "/" - try: - exp = int(ck[4]) - if exp > 0: - c["expiry"] = exp - except Exception: - pass + exp = _safe_expiry(ck[4]) + if exp > 0: + c["expiry"] = exp c["secure"] = str(ck[5]).upper() == "TRUE" try: driver.add_cookie(c) @@ -271,8 +400,8 @@ def export(): 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": int(r[4]) if r[4] and int(r[4]) > 0 else None, - "secure": r[5].upper() == "TRUE", "httpOnly": False} for r in rows] + "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"}) @@ -287,7 +416,7 @@ def export_ns(): 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] + 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"})