Initial commit: HA Voice Control MCP server
This commit is contained in:
@@ -0,0 +1,284 @@
|
||||
# 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()
|
||||
Reference in New Issue
Block a user