Gan, Jimmy
a0ffa41c9e
- Add auth.py: OAuth credential manager with auto-refresh - Rewrite fetcher.py: use googleapiclient subscriptions/search - Add google-auth, google-api-python-client dependencies - Update docker-compose for config volume mount - Add .gitignore for secrets (client_secret.json, token.pickle) - Live test: 198 videos ingested from 229 channels
98 lines
3.4 KiB
Python
98 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Export YouTube cookies from Chrome. Run: ytc"""
|
|
|
|
import os, sys, sqlite3, subprocess, hashlib, shutil, tempfile
|
|
from pathlib import Path
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
OUTPUT = REPO_ROOT / "config" / "cookies.txt"
|
|
COOKIES_DB = Path.home() / "Library/Application Support/Google/Chrome/Default/Cookies"
|
|
|
|
SALT = b'saltysalt'
|
|
KEYLEN = 16 # AES-128
|
|
PBKDF2_ITERS = 1003
|
|
|
|
|
|
def get_key():
|
|
r = subprocess.run(
|
|
['security', 'find-generic-password', '-w', '-s', 'Chrome Safe Storage'],
|
|
capture_output=True, text=True, timeout=30
|
|
)
|
|
if r.returncode != 0:
|
|
print("ERROR: Keychain access failed. Approve the dialog.", file=sys.stderr)
|
|
sys.exit(1)
|
|
return hashlib.pbkdf2_hmac('sha1', r.stdout.strip().encode(), SALT, PBKDF2_ITERS, KEYLEN)
|
|
|
|
|
|
def decrypt(blob, key):
|
|
"""Decrypt Chrome v10 encrypted value. Returns str or None."""
|
|
if not blob or not blob.startswith(b'v10'):
|
|
return None
|
|
try:
|
|
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
|
from cryptography.hazmat.primitives import padding
|
|
iv = blob[3:19]
|
|
ct = blob[19:]
|
|
c = Cipher(algorithms.AES(key), modes.CBC(iv))
|
|
d = c.decryptor()
|
|
pt = d.update(ct) + d.finalize()
|
|
u = padding.PKCS7(128).unpadder()
|
|
return (u.update(pt) + u.finalize()).decode('utf-8', errors='replace')
|
|
except Exception as e:
|
|
return None
|
|
|
|
|
|
def main():
|
|
print("Getting Chrome key from Keychain... (approve dialog)")
|
|
key = get_key()
|
|
|
|
# Copy DB to avoid lock
|
|
with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as f:
|
|
tmp = f.name
|
|
shutil.copy2(COOKIES_DB, tmp)
|
|
|
|
conn = sqlite3.connect(tmp)
|
|
conn.text_factory = bytes
|
|
rows = conn.execute(
|
|
"SELECT host_key, name, encrypted_value, value, path, expires_utc, is_secure, is_httponly "
|
|
"FROM cookies WHERE host_key LIKE '%youtube.com' ORDER BY name"
|
|
).fetchall()
|
|
conn.close()
|
|
os.unlink(tmp)
|
|
|
|
print(f"Found {len(rows)} YouTube cookies")
|
|
|
|
OUTPUT.parent.mkdir(parents=True, exist_ok=True)
|
|
ok, fail, skipped = 0, 0, 0
|
|
|
|
with open(OUTPUT, 'w') as f:
|
|
f.write("# Netscape HTTP Cookie File\n# Extracted from Chrome\n\n")
|
|
for hk, name, enc_val, raw_val, path, expires, secure, httponly in rows:
|
|
val = decrypt(enc_val, key)
|
|
if val is None and raw_val:
|
|
val = raw_val.decode('utf-8', errors='replace')
|
|
if not val:
|
|
n = name.decode() if isinstance(name, bytes) else name
|
|
print(f" SKIP: {n} (decrypt failed, no raw value)")
|
|
skipped += 1
|
|
continue
|
|
|
|
d = hk.decode() if isinstance(hk, bytes) else hk
|
|
n = name.decode() if isinstance(name, bytes) else name
|
|
p = path.decode() if isinstance(path, bytes) else path
|
|
flag = "TRUE" if d.startswith('.') else "FALSE"
|
|
s = "TRUE" if secure else "FALSE"
|
|
e = int(expires / 1_000_000) if expires and expires > 0 else 0
|
|
|
|
if httponly:
|
|
f.write(f"#HttpOnly_{d}\t{flag}\t{p}\t{s}\t{e}\t{n}\t{val}\n")
|
|
f.write(f"{d}\t{flag}\t{p}\t{s}\t{e}\t{n}\t{val}\n")
|
|
ok += 1
|
|
|
|
print(f"\nExported: {ok} Failed: {fail} Skipped: {skipped}")
|
|
print(f"Output: {OUTPUT} ({OUTPUT.stat().st_size} bytes)")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|