From 8d341e0d614b2d363e338467ed6539eee204de90 Mon Sep 17 00:00:00 2001 From: admin Date: Wed, 6 Aug 2025 17:08:09 +0000 Subject: [PATCH] Add pwWinSync.py --- pwWinSync.py | 223 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 223 insertions(+) create mode 100644 pwWinSync.py diff --git a/pwWinSync.py b/pwWinSync.py new file mode 100644 index 0000000..599c4a1 --- /dev/null +++ b/pwWinSync.py @@ -0,0 +1,223 @@ +import win32process +import win32pipe +import win32file +import win32event +import win32con +import pywintypes +import json +import sys +import time +from datetime import datetime +from pykeepass import PyKeePass, exceptions as kp_exceptions + +# ========== CONFIGURE THESE VALUES ========== +KEEPASS_DB = r"C:\path\to\your\database.kdbx" +KEEPASS_PASS = "your_kdbx_password" +BW_EMAIL = "your_vaultwarden_email" +BW_PASS = "your_vaultwarden_password" +BW_SERVER_URL = "https://your.vaultwarden.instance" +# ============================================ + +def error(msg): + print(f"[ERROR] {msg}", file=sys.stderr) + sys.exit(1) + +def run_command(cmd_list): + """ + Run a CLI command using pywin32, returning stdout, stderr, and exit code. + """ + # Make a suitable command string for Windows + cmd = ' '.join(f'"{c}"' if ' ' in c else c for c in cmd_list) + # Create pipes for stdout and stderr + read_out, write_out = win32pipe.CreatePipe(None, 0) + read_err, write_err = win32pipe.CreatePipe(None, 0) + si = win32process.STARTUPINFO() + si.dwFlags |= win32con.STARTF_USESTDHANDLES + si.hStdOutput = write_out + si.hStdError = write_err + + try: + hProcess, hThread, _, _ = win32process.CreateProcess( + None, cmd, None, None, True, 0, None, None, si + ) + + win32file.CloseHandle(write_out) + win32file.CloseHandle(write_err) + + win32event.WaitForSingleObject(hProcess, win32event.INFINITE) + + # Read stdout + stdout = b"" + while True: + try: + chunk = win32file.ReadFile(read_out, 4096)[1] + if not chunk: break + stdout += chunk + except pywintypes.error: + break + # Read stderr + stderr = b"" + while True: + try: + chunk = win32file.ReadFile(read_err, 4096)[1] + if not chunk: break + stderr += chunk + except pywintypes.error: + break + + exit_code = win32process.GetExitCodeProcess(hProcess) + + win32file.CloseHandle(read_out) + win32file.CloseHandle(read_err) + win32process.CloseHandle(hProcess) + win32process.CloseHandle(hThread) + return stdout.decode('utf-8', errors='replace'), stderr.decode('utf-8', errors='replace'), exit_code + + except pywintypes.error as e: + return "", str(e), -1 + +# Date/time helpers +def keepass_lastmod(entry): + return time.mktime(entry.times.lastmod.timetuple()) if entry.times else 0 + +def vaultwarden_lastmod(item): + if not item.get('revisionDate'): return 0 + try: + dt = datetime.strptime(item['revisionDate'], "%Y-%m-%dT%H:%M:%S.%fZ") + return dt.timestamp() + except Exception: + return 0 + +# Bitwarden CLI via win32 process +def bw_login(): + stdout, stderr, exitcode = run_command(['bw', 'login', BW_EMAIL, BW_PASS, '--raw', '--server', BW_SERVER_URL]) + if exitcode != 0 or not stdout.strip(): + error(f"Bitwarden login failed: {stderr}") + return stdout.strip() + +def bw_sync(session): + _, stderr, exitcode = run_command(['bw', 'sync', '--session', session, '--server', BW_SERVER_URL]) + if exitcode != 0: + error(f"Bitwarden sync failed: {stderr}") + +def bw_export(session): + stdout, stderr, exitcode = run_command([ + 'bw', 'export', '--format', 'json', '--session', session, '--server', BW_SERVER_URL + ]) + if exitcode != 0 or not stdout.strip(): + error(f"Bitwarden export failed: {stderr}") + return json.loads(stdout) + +def bw_update_item(updated_item, session): + _, stderr, exitcode = run_command([ + 'bw', 'edit', 'item', updated_item['id'], json.dumps(updated_item), '--session', session, '--server', BW_SERVER_URL + ]) + if exitcode != 0: + error(f"Failed to update item in Bitwarden: {stderr}") + +def bw_create_item(item, session): + payload = { + "type": 1, + "name": item['title'], + "login": { + "username": item['username'], + "password": item['password'], + "uris": [{"uri": item['url']}] if item['url'] else [] + } + } + _, stderr, exitcode = run_command([ + 'bw', 'create', 'item', json.dumps(payload), '--session', session, '--server', BW_SERVER_URL + ]) + if exitcode != 0: + error(f"Failed to add item to Bitwarden: {stderr}") + +# KeePass DB functions +def load_keepass(): + try: + kp = PyKeePass(KEEPASS_DB, password=KEEPASS_PASS) + return kp + except kp_exceptions.CredentialsError: + error("Invalid KeePass credentials or file not found.") + except Exception as e: + error(f"Could not open KeePass database: {e}") + +def add_keepass_entry(kp, title, username, password, url): + try: + kp.add_entry(kp.root_group, title, username, password, url=url) + except Exception as e: + error(f"Failed to add entry to KeePass: {e}") + +def update_keepass_entry(entry, password, url): + changed = False + if entry.password != password: + entry.password = password + changed = True + if entry.url != url: + entry.url = url + changed = True + if changed: + entry.touch() + +# Main sync and conflict logic +def main(): + kp = load_keepass() + session = bw_login() + bw_sync(session) + bw_data = bw_export(session) + + # Index entries for matching (Title+Username) + keepass_entries = {(e.title, e.username): e for e in kp.entries if e.username} + bw_entries = {} + for item in bw_data['items']: + if item['type'] == 1: # Only Login items + username = item['login'].get('username') if item['login'] else None + bw_entries[(item['name'], username)] = item + + for key in set(keepass_entries.keys()).union(bw_entries.keys()): + kp_entry = keepass_entries.get(key) + bw_item = bw_entries.get(key) + + if kp_entry and not bw_item: + print(f"Adding KeePass entry to Vaultwarden: {key}") + bw_create_item({ + "title": kp_entry.title, + "username": kp_entry.username, + "password": kp_entry.password, + "url": kp_entry.url + }, session) + + elif bw_item and not kp_entry and bw_item['login'].get('password'): + print(f"Adding Vaultwarden entry to KeePass: {key}") + add_keepass_entry( + kp, + bw_item['name'], + bw_item['login'].get('username', ''), + bw_item['login'].get('password', ''), + (bw_item['login'].get('uris') or [{}])[0].get('uri', '') + ) + + elif kp_entry and bw_item: + kp_lastmod = keepass_lastmod(kp_entry) + bw_lastmod = vaultwarden_lastmod(bw_item) + if kp_lastmod > bw_lastmod + 1: + print(f"Conflict ({key}): updating Vaultwarden from KeePass (KeePass modified later).") + bw_item['login']['password'] = kp_entry.password + bw_item['login']['uris'] = [{"uri": kp_entry.url}] if kp_entry.url else [] + bw_update_item(bw_item, session) + elif bw_lastmod > kp_lastmod + 1: + print(f"Conflict ({key}): updating KeePass from Vaultwarden (Vaultwarden modified later).") + update_keepass_entry( + kp_entry, + bw_item['login'].get('password', ''), + (bw_item['login'].get('uris') or [{}])[0].get('uri', '') + ) + # else: No significant update + + try: + kp.save() + print("Synchronization complete, with conflict resolution.") + except Exception as e: + error(f"Saving KeePass database failed: {e}") + +if __name__ == '__main__': + main()