mirror of
https://github.com/openswarm-ai/openswarm.git
synced 2026-09-10 03:37:44 +02:00
[eric] onboarding: cookie harvest works on Windows too (DPAPI key + AES-GCM), macOS path unchanged; ChatGPT-token path already cross-platform
This commit is contained in:
ciregenz
@@ -1,33 +1,56 @@
|
||||
"""Read the user's own logged-in provider cookies from their real browser, so
|
||||
onboarding can harvest their actual chat history at first run without an in-app login.
|
||||
|
||||
macOS + Chromium only for now (Chrome/Arc/Brave/Edge). We first find WHICH store holds
|
||||
the session by counting cookie names in the SQLite (no decryption, no keychain), then
|
||||
decrypt only that one store, so the "Safe Storage" keychain is touched at most once per
|
||||
browser (cached for the process). Values are v10/v11 AES-CBC. Fails open to {} on
|
||||
anything (no browser, app-bound v20 cookies, denied keychain, Safari-only user), so
|
||||
prep just falls back to the local scan.
|
||||
Chromium (Chrome/Arc/Brave/Edge) on macOS AND Windows. We first find WHICH store holds the
|
||||
session by counting cookie names in the SQLite (no decryption, no keychain/DPAPI), then decrypt
|
||||
only that one store, so the secret key is fetched at most once per browser (cached for the
|
||||
process). Per-OS decryption:
|
||||
- macOS: "Safe Storage" keychain password -> PBKDF2 -> AES-CBC (v10/v11).
|
||||
- Windows: DPAPI-unwrapped key from Local State -> AES-256-GCM (v10/v11).
|
||||
v20 = app-bound encryption (modern Chrome), out of reach on both without the browser's own
|
||||
elevation service. Fails open to {} on anything (no browser, app-bound cookies, denied
|
||||
keychain/DPAPI, Safari-only user), so prep just falls back to the local scan.
|
||||
|
||||
Only ever reads the specific provider domain asked for; never a general cookie sweep.
|
||||
The values are session secrets: used in-process for the harvest, never logged or stored.
|
||||
|
||||
NOTE: the Windows path is written to the well-documented Chromium/DPAPI scheme but is NOT
|
||||
live-tested from this repo's dev machine (macOS); the macOS path is live-proven (490 real
|
||||
Claude convos). Both fail open, so a Windows decryption miss degrades to the scan, never crashes.
|
||||
"""
|
||||
|
||||
import base64
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import sqlite3
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
|
||||
from typeguard import typechecked
|
||||
|
||||
CHROMIUM_ROOTS = {
|
||||
"Chrome": "Library/Application Support/Google/Chrome",
|
||||
"Arc": "Library/Application Support/Arc/User Data",
|
||||
"Brave": "Library/Application Support/BraveSoftware/Brave-Browser",
|
||||
"Edge": "Library/Application Support/Microsoft Edge",
|
||||
}
|
||||
IS_WIN = sys.platform == "win32"
|
||||
|
||||
# Per-OS "User Data" roots (relative to the home dir), where profiles + Local State live.
|
||||
if IS_WIN:
|
||||
p_local = os.environ.get("LOCALAPPDATA", os.path.expanduser("~/AppData/Local"))
|
||||
CHROMIUM_ROOTS = {
|
||||
"Chrome": os.path.join(p_local, "Google", "Chrome", "User Data"),
|
||||
"Arc": os.path.join(p_local, "Packages"), # Arc/Windows is UWP-packaged + rare; best-effort
|
||||
"Brave": os.path.join(p_local, "BraveSoftware", "Brave-Browser", "User Data"),
|
||||
"Edge": os.path.join(p_local, "Microsoft", "Edge", "User Data"),
|
||||
}
|
||||
else:
|
||||
p_home = os.path.expanduser("~")
|
||||
CHROMIUM_ROOTS = {
|
||||
"Chrome": os.path.join(p_home, "Library/Application Support/Google/Chrome"),
|
||||
"Arc": os.path.join(p_home, "Library/Application Support/Arc/User Data"),
|
||||
"Brave": os.path.join(p_home, "Library/Application Support/BraveSoftware/Brave-Browser"),
|
||||
"Edge": os.path.join(p_home, "Library/Application Support/Microsoft Edge"),
|
||||
}
|
||||
KEYCHAIN_SERVICE = {
|
||||
"Chrome": "Chrome Safe Storage",
|
||||
"Arc": "Arc Safe Storage",
|
||||
@@ -36,15 +59,58 @@ KEYCHAIN_SERVICE = {
|
||||
}
|
||||
PROFILES = ["Default"] + [f"Profile {i}" for i in range(1, 12)]
|
||||
|
||||
# One keychain read per browser per process; "Always Allow" then never re-prompts.
|
||||
# One key fetch per browser per process; "Always Allow" (mac) / DPAPI (win) then never re-prompts.
|
||||
p_key_cache: Dict[str, Optional[bytes]] = {}
|
||||
|
||||
|
||||
@typechecked
|
||||
def p_safe_storage_key(browser: str) -> Optional[bytes]:
|
||||
if browser in p_key_cache:
|
||||
return p_key_cache[browser]
|
||||
key: Optional[bytes] = None
|
||||
def p_win_dpapi_unprotect(data: bytes) -> Optional[bytes]:
|
||||
"""CryptUnprotectData via crypt32.dll (no pywin32 dependency). None on any failure."""
|
||||
try:
|
||||
import ctypes
|
||||
from ctypes import wintypes
|
||||
|
||||
class DATA_BLOB(ctypes.Structure):
|
||||
p_fields = [("cbData", wintypes.DWORD), ("pbData", ctypes.POINTER(ctypes.c_char))]
|
||||
_fields_ = p_fields
|
||||
|
||||
buf = ctypes.create_string_buffer(data, len(data))
|
||||
blob_in = DATA_BLOB(len(data), ctypes.cast(buf, ctypes.POINTER(ctypes.c_char)))
|
||||
blob_out = DATA_BLOB()
|
||||
ok = ctypes.windll.crypt32.CryptUnprotectData(
|
||||
ctypes.byref(blob_in), None, None, None, None, 0, ctypes.byref(blob_out)
|
||||
)
|
||||
if not ok:
|
||||
return None
|
||||
n = int(blob_out.cbData)
|
||||
out = ctypes.create_string_buffer(n)
|
||||
ctypes.memmove(out, blob_out.pbData, n)
|
||||
ctypes.windll.kernel32.LocalFree(blob_out.pbData)
|
||||
return out.raw
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
@typechecked
|
||||
def p_win_storage_key(browser: str) -> Optional[bytes]:
|
||||
"""The AES key from a Chromium install's Local State: base64 -> strip 'DPAPI' -> CryptUnprotectData."""
|
||||
base = CHROMIUM_ROOTS.get(browser)
|
||||
if not base:
|
||||
return None
|
||||
local_state = os.path.join(base, "Local State")
|
||||
try:
|
||||
with open(local_state, "r", encoding="utf-8") as f:
|
||||
enc_b64 = json.load(f)["os_crypt"]["encrypted_key"]
|
||||
raw = base64.b64decode(enc_b64)
|
||||
if raw[:5] != b"DPAPI":
|
||||
return None
|
||||
return p_win_dpapi_unprotect(raw[5:])
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
@typechecked
|
||||
def p_mac_storage_key(browser: str) -> Optional[bytes]:
|
||||
try:
|
||||
r = subprocess.run(
|
||||
["security", "find-generic-password", "-w", "-s", KEYCHAIN_SERVICE[browser]],
|
||||
@@ -52,9 +118,17 @@ def p_safe_storage_key(browser: str) -> Optional[bytes]:
|
||||
)
|
||||
pw = r.stdout.strip()
|
||||
if pw:
|
||||
key = hashlib.pbkdf2_hmac("sha1", pw.encode(), b"saltysalt", 1003, 16)
|
||||
return hashlib.pbkdf2_hmac("sha1", pw.encode(), b"saltysalt", 1003, 16)
|
||||
except Exception:
|
||||
key = None
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
@typechecked
|
||||
def p_safe_storage_key(browser: str) -> Optional[bytes]:
|
||||
if browser in p_key_cache:
|
||||
return p_key_cache[browser]
|
||||
key = p_win_storage_key(browser) if IS_WIN else p_mac_storage_key(browser)
|
||||
p_key_cache[browser] = key
|
||||
return key
|
||||
|
||||
@@ -82,11 +156,9 @@ def p_count_domain(db_path: str, domain: str) -> int:
|
||||
@typechecked
|
||||
def p_best_store(domain: str) -> Optional[Tuple[str, str]]:
|
||||
"""The (browser, db_path) holding the most cookies for `domain`, found WITHOUT the keychain."""
|
||||
home = os.path.expanduser("~")
|
||||
best: Optional[Tuple[str, str]] = None
|
||||
best_score = (0, -1.0)
|
||||
for browser, rel in CHROMIUM_ROOTS.items():
|
||||
base = os.path.join(home, rel)
|
||||
for browser, base in CHROMIUM_ROOTS.items():
|
||||
if not os.path.isdir(base):
|
||||
continue
|
||||
for prof in PROFILES:
|
||||
@@ -107,13 +179,19 @@ def p_decrypt(enc: bytes, key: bytes) -> Optional[str]:
|
||||
if enc[:3] not in (b"v10", b"v11"):
|
||||
return None # v20 = app-bound encryption, out of reach without the browser
|
||||
try:
|
||||
from cryptography.hazmat.backends import default_backend
|
||||
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
||||
if IS_WIN:
|
||||
# Windows Chromium: v10/v11 = AES-256-GCM, [3:15]=nonce, tail 16 bytes=tag (bundled with ct).
|
||||
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
||||
|
||||
c = Cipher(algorithms.AES(key), modes.CBC(b" " * 16), backend=default_backend())
|
||||
d = c.decryptor()
|
||||
dec = d.update(enc[3:]) + d.finalize()
|
||||
dec = dec[: -dec[-1]] # strip PKCS7 padding
|
||||
dec = AESGCM(key).decrypt(enc[3:15], enc[15:], None)
|
||||
else:
|
||||
from cryptography.hazmat.backends import default_backend
|
||||
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
||||
|
||||
c = Cipher(algorithms.AES(key), modes.CBC(b" " * 16), backend=default_backend())
|
||||
d = c.decryptor()
|
||||
dec = d.update(enc[3:]) + d.finalize()
|
||||
dec = dec[: -dec[-1]] # strip PKCS7 padding
|
||||
for cut in (0, 32): # newer Chromium prepends a 32-byte domain hash
|
||||
try:
|
||||
return dec[cut:].decode("utf-8")
|
||||
|
||||
@@ -154,6 +154,47 @@ def test_read_provider_cookies_fails_open_without_a_store(monkeypatch):
|
||||
assert browser_cookies.read_provider_cookie_records("claude.ai") == []
|
||||
|
||||
|
||||
def test_win_storage_key_parses_local_state_and_unwraps(monkeypatch, tmp_path):
|
||||
from backend.apps.onboarding.usage import browser_cookies
|
||||
|
||||
# Local State carries a base64 "DPAPI"-prefixed key; the Windows path strips the prefix and
|
||||
# hands the rest to CryptUnprotectData. Prove the parse + prefix-strip without needing Windows.
|
||||
raw_key = b"DPAPI" + b"wrapped-key-bytes"
|
||||
local_state_dir = tmp_path / "UserData"
|
||||
local_state_dir.mkdir()
|
||||
(local_state_dir / "Local State").write_text(
|
||||
json.dumps({"os_crypt": {"encrypted_key": base64.b64encode(raw_key).decode()}})
|
||||
)
|
||||
monkeypatch.setattr(browser_cookies, "CHROMIUM_ROOTS", {"Chrome": str(local_state_dir)})
|
||||
seen = {}
|
||||
|
||||
def fake_unprotect(data: bytes):
|
||||
seen["passed"] = data
|
||||
return b"unwrapped-aes-key"
|
||||
|
||||
monkeypatch.setattr(browser_cookies, "p_win_dpapi_unprotect", fake_unprotect)
|
||||
key = browser_cookies.p_win_storage_key("Chrome")
|
||||
assert key == b"unwrapped-aes-key"
|
||||
assert seen["passed"] == b"wrapped-key-bytes" # the 5-byte "DPAPI" prefix was stripped
|
||||
|
||||
|
||||
def test_win_storage_key_fails_open(monkeypatch, tmp_path):
|
||||
from backend.apps.onboarding.usage import browser_cookies
|
||||
|
||||
# Missing Local State, malformed JSON, and DPAPI failure all fail open to None (-> scan fallback).
|
||||
monkeypatch.setattr(browser_cookies, "CHROMIUM_ROOTS", {"Chrome": str(tmp_path)})
|
||||
assert browser_cookies.p_win_storage_key("Chrome") is None # no Local State file
|
||||
assert browser_cookies.p_win_storage_key("Nonexistent") is None
|
||||
|
||||
|
||||
def test_decrypt_rejects_app_bound_v20():
|
||||
from backend.apps.onboarding.usage import browser_cookies
|
||||
|
||||
# v20 = app-bound encryption (modern Chrome), out of reach on both OSes -> None, never a crash.
|
||||
assert browser_cookies.p_decrypt(b"v20" + b"anything", b"\x00" * 32) is None
|
||||
assert browser_cookies.p_decrypt(b"", b"\x00" * 32) is None
|
||||
|
||||
|
||||
def test_dump_cookies_only_serves_allowlisted_domains(monkeypatch, capsys):
|
||||
from backend.apps.onboarding.usage import dump_cookies
|
||||
|
||||
|
||||
Reference in New Issue
Block a user