Upload files to "/"
This commit is contained in:
@@ -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()
|
||||
Reference in New Issue
Block a user