commit 9af0933c7e4db1233d7d92e40b965549838ecbeb Author: admin Date: Mon Aug 4 17:19:49 2025 +0000 Upload files to "/" diff --git a/pwSync.py b/pwSync.py new file mode 100644 index 0000000..013e399 --- /dev/null +++ b/pwSync.py @@ -0,0 +1,186 @@ +import subprocess +import json +from pykeepass import PyKeePass, exceptions as kp_exceptions +import os +import sys +import time +from datetime import datetime + +# CONFIGURATION +KEEPASS_DB = r"C:\path\to\your\database.kdbx" +KEEPASS_PASS = "your_kdbx_password" +BW_EMAIL = "your_vaultwarden_email" +BW_PASS = "your_vaultwarden_password" + +def error(msg): + print(f"[ERROR] {msg}", file=sys.stderr) + sys.exit(1) + +# --- Date/time utilities --- # +def keepass_lastmod(entry): + """Returns POSIX timestamp for last modified time of a KeePass entry.""" + return time.mktime(entry.times.lastmod.timetuple()) if entry.times else 0 + +def vaultwarden_lastmod(item): + """Returns POSIX timestamp for Vaultwarden's last_edited.""" + if not item.get('revisionDate'): + return 0 + # Example: "2023-08-03T13:14:15.567Z" + try: + dt = datetime.strptime(item['revisionDate'], "%Y-%m-%dT%H:%M:%S.%fZ") + return dt.timestamp() + except: + return 0 + +# --- Vaultwarden/Bitwarden CLI functions --- # +def bw_login(): + try: + result = subprocess.run(['bw', 'login', BW_EMAIL, BW_PASS, '--raw'], capture_output=True, text=True, check=True) + session = result.stdout.strip() + if not session: + error("Failed to obtain Bitwarden session.") + return session + except Exception as e: + error(f"Bitwarden login failed: {e}") + +def bw_sync(session): + try: + subprocess.run(['bw', 'sync', '--session', session], check=True) + except Exception as e: + error(f"Bitwarden sync failed: {e}") + +def bw_export(session): + try: + result = subprocess.run(['bw', 'export', '--format', 'json', '--session', session], capture_output=True, text=True, check=True) + data = json.loads(result.stdout) + return data + except Exception as e: + error(f"Bitwarden export failed: {e}") + +def bw_update_item(updated_item, session): + try: + subprocess.run(['bw', 'edit', 'item', updated_item['id'], json.dumps(updated_item), '--session', session], + check=True, capture_output=True) + except Exception as e: + error(f"Failed to update item in Bitwarden: {e}") + +def bw_create_item(item, session): + payload = { + "type": 1, # 1 = Login + "name": item['title'], + "login": { + "username": item['username'], + "password": item['password'], + "uris": [{"uri": item['url']} if item['url'] else {}] + } + } + try: + subprocess.run(['bw', 'create', 'item', json.dumps(payload), '--session', session], check=True) + except Exception as e: + error(f"Failed to add item to Bitwarden: {e}") + +# --- KeePass 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 find_keepass_entry(kp, title, username): + for entry in kp.entries: + if entry.title == title and entry.username == username: + return entry + return None + +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() + +# --- Sync logic with conflict resolution --- # +def main(): + kp = load_keepass() + session = bw_login() + bw_sync(session) + bw_data = bw_export(session) + + 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: + username = item['login'].get('username') if item['login'] else None + bw_entries[(item['name'], username)] = item + + # 1. Synchronization and conflict resolution + for key in set(keepass_entries.keys()).union(bw_entries.keys()): + kp_entry = keepass_entries.get(key) + bw_item = bw_entries.get(key) + + # Entry only in KeePass + 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) + + # Entry only in Vaultwarden + elif bw_item and not kp_entry and bw_item['login']['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', '') + ) + + # Entry in both: conflict resolution + elif kp_entry and bw_item: + kp_lastmod = keepass_lastmod(kp_entry) + bw_lastmod = vaultwarden_lastmod(bw_item) + # Compare fields + if kp_lastmod > bw_lastmod + 1: + # KeePass is newer + print(f"Conflict ({key}): updating Vaultwarden from KeePass (KeePass modified later).") + # Update Bitwarden's entry + 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: + # Vaultwarden is newer + 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: + pass # No significant difference + + # Save changes + try: + kp.save() + print("Synchronization complete, with conflict resolution.") + except Exception as e: + error(f"Saving KeePass database failed: {e}") + +if __name__ == '__main__': + main()