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()