pwsync/pwSync.py

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