469 lines
17 KiB
Python
469 lines
17 KiB
Python
"""
|
|
Decky Vaultwarden - Bitwarden/Vaultwarden password manager plugin for Decky Loader.
|
|
|
|
Main entry point for the Python backend. Provides API routes for the frontend.
|
|
"""
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
from typing import Optional
|
|
|
|
plugin_dir = os.path.dirname(os.path.abspath(__file__))
|
|
py_modules_dir = os.path.join(plugin_dir, "py_modules")
|
|
|
|
sys.path.insert(0, py_modules_dir)
|
|
sys.path.insert(0, plugin_dir)
|
|
|
|
if "SSL_CERT_FILE" not in os.environ:
|
|
for ca_path in ["/etc/ssl/certs/ca-certificates.crt", "/etc/ssl/certs/ca-bundle.crt"]:
|
|
if os.path.exists(ca_path):
|
|
os.environ["SSL_CERT_FILE"] = ca_path
|
|
break
|
|
|
|
def _ensure_deps():
|
|
try:
|
|
import cryptography # noqa: F401
|
|
return
|
|
except ImportError:
|
|
pass
|
|
|
|
import importlib.util
|
|
cffi_path = os.path.join(py_modules_dir, "_cffi_backend.cpython-311-x86_64-linux-gnu.so")
|
|
if os.path.exists(cffi_path) and "_cffi_backend" not in sys.modules:
|
|
spec = importlib.util.spec_from_file_location("_cffi_backend", cffi_path)
|
|
mod = importlib.util.module_from_spec(spec)
|
|
sys.modules["_cffi_backend"] = mod
|
|
spec.loader.exec_module(mod)
|
|
|
|
_ensure_deps()
|
|
|
|
import decky_plugin
|
|
|
|
|
|
class Plugin:
|
|
"""Main plugin class for Decky Vaultwarden."""
|
|
|
|
def __init__(self):
|
|
self.client = None
|
|
self.client_class = None
|
|
self.pin_crypto = None
|
|
self.crypto = None
|
|
self._vault_data = None
|
|
self._decrypted_vault = None
|
|
self._master_key = None
|
|
self._enc_key = None
|
|
self._mac_key = None
|
|
self._session_active = False
|
|
self._settings_path = os.path.join(
|
|
os.path.expanduser("~"), ".config", "decky-vaultwarden"
|
|
)
|
|
self._pin_attempts = 0
|
|
self._MAX_PIN_ATTEMPTS = 5
|
|
self._deps_ready = False
|
|
|
|
# ===== Lifecycle Methods =====
|
|
|
|
async def _main(self):
|
|
"""Called when the plugin is loaded."""
|
|
from bitwarden_client import BitwardenClient
|
|
from crypto import BitwardenCrypto, PinCrypto
|
|
|
|
self.client_class = BitwardenClient
|
|
self.pin_crypto = PinCrypto()
|
|
self.crypto = BitwardenCrypto()
|
|
self._deps_ready = True
|
|
decky_plugin.logger.info("Decky Vaultwarden plugin loaded")
|
|
os.makedirs(self._settings_path, exist_ok=True)
|
|
|
|
async def _unload(self):
|
|
"""Called when the plugin is unloaded."""
|
|
await self.lock_vault()
|
|
decky_plugin.logger.info("Decky Vaultwarden plugin unloaded")
|
|
|
|
async def _migration(self):
|
|
"""Called when plugin version changes."""
|
|
pass
|
|
|
|
# ===== Authentication =====
|
|
|
|
async def login_password(
|
|
self, server_url: str, email: str, password: str, two_factor_token: Optional[str] = None
|
|
) -> dict:
|
|
"""Login with email and master password."""
|
|
try:
|
|
self.client = self.client_class(server_url)
|
|
result = await self.client.login_password(email, password, two_factor_token)
|
|
|
|
if result.get("two_factor_required"):
|
|
return {"two_factor_required": True}
|
|
|
|
if result.get("success"):
|
|
self._master_key = result.get("master_key")
|
|
self._enc_key = result.get("enc_key")
|
|
self._mac_key = result.get("mac_key")
|
|
self._session_active = True
|
|
self._save_settings({
|
|
"server_url": server_url,
|
|
"email": email,
|
|
"kdf_info": self.client.get_kdf_info(),
|
|
})
|
|
|
|
# Sync and decrypt vault immediately if we have keys
|
|
if self._enc_key and self._mac_key:
|
|
try:
|
|
sync_data = await self.client.sync_vault()
|
|
self._vault_data = sync_data
|
|
self._decrypted_vault = self.client.decrypt_vault(
|
|
sync_data, self._enc_key, self._mac_key
|
|
)
|
|
return {"success": True, "needs_unlock": False}
|
|
except Exception as e:
|
|
decky_plugin.logger.error(f"Vault sync/decrypt failed: {e}")
|
|
return {"success": True, "needs_unlock": False}
|
|
|
|
return {"success": True, "needs_unlock": result.get("enc_key") is None}
|
|
|
|
return {"success": False, "error": "Login failed"}
|
|
|
|
except Exception as e:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
async def login_api_key(
|
|
self, server_url: str, client_id: str, client_secret: str, email: str
|
|
) -> dict:
|
|
"""Login with API key."""
|
|
try:
|
|
self.client = self.client_class(server_url)
|
|
result = await self.client.login_api_key(client_id, client_secret, email)
|
|
|
|
if result.get("success"):
|
|
self._session_active = True
|
|
self._save_settings({
|
|
"server_url": server_url,
|
|
"email": email,
|
|
"auth_method": "api_key",
|
|
"kdf_info": self.client.get_kdf_info(),
|
|
})
|
|
return {
|
|
"success": True,
|
|
"needs_master_password": True,
|
|
"message": "API key authenticated. Enter master password to unlock vault.",
|
|
}
|
|
|
|
return {"success": False, "error": result.get("error", "API key login failed")}
|
|
|
|
except Exception as e:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
async def unlock_with_master_password(self, password: str) -> dict:
|
|
"""Unlock vault with master password (for API key users or fresh login)."""
|
|
try:
|
|
if not self.client:
|
|
return {"success": False, "error": "Not authenticated. Login first."}
|
|
|
|
result = await self.client.unlock_with_master_password(password)
|
|
self._master_key = result.get("master_key")
|
|
|
|
# Sync and decrypt vault
|
|
sync_data = await self.client.sync_vault()
|
|
self._vault_data = sync_data
|
|
|
|
# Get encryption keys from login or derive them
|
|
if not self._enc_key and self._master_key:
|
|
enc_user_key = sync_data.get("profile", {}).get("key")
|
|
if enc_user_key:
|
|
self._enc_key, self._mac_key = self.crypto.decrypt_user_key(
|
|
enc_user_key, self._master_key
|
|
)
|
|
|
|
if self._enc_key and self._mac_key:
|
|
self._decrypted_vault = self.client.decrypt_vault(
|
|
sync_data, self._enc_key, self._mac_key
|
|
)
|
|
return {"success": True, "vault_size": len(self._decrypted_vault.get("ciphers", []))}
|
|
|
|
return {"success": False, "error": "Could not derive encryption keys"}
|
|
|
|
except Exception as e:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
# ===== Vault Operations =====
|
|
|
|
async def get_vault_items(self) -> dict:
|
|
"""Get decrypted vault items."""
|
|
if not self._decrypted_vault:
|
|
return {"success": False, "error": "Vault not unlocked"}
|
|
return {"success": True, "data": self._decrypted_vault}
|
|
|
|
async def get_folders(self) -> dict:
|
|
"""Get decrypted folders."""
|
|
if not self._decrypted_vault:
|
|
return {"success": False, "error": "Vault not unlocked"}
|
|
return {
|
|
"success": True,
|
|
"folders": self._decrypted_vault.get("folders", []),
|
|
}
|
|
|
|
async def search_vault(self, query: str) -> dict:
|
|
"""Search vault items by name, username, or URI."""
|
|
if not self._decrypted_vault:
|
|
return {"success": False, "error": "Vault not unlocked"}
|
|
|
|
query_lower = query.lower()
|
|
results = []
|
|
|
|
for cipher in self._decrypted_vault.get("ciphers", []):
|
|
if cipher.get("error"):
|
|
continue
|
|
|
|
# Search in name
|
|
name = cipher.get("name", "")
|
|
if query_lower in name.lower():
|
|
results.append(cipher)
|
|
continue
|
|
|
|
# Search in login fields
|
|
login = cipher.get("login", {})
|
|
username = login.get("username", "") or ""
|
|
if query_lower in username.lower():
|
|
results.append(cipher)
|
|
continue
|
|
|
|
# Search in URIs
|
|
for uri in login.get("uris", []):
|
|
uri_str = uri.get("uri", "") or ""
|
|
if query_lower in uri_str.lower():
|
|
results.append(cipher)
|
|
break
|
|
|
|
return {"success": True, "results": results}
|
|
|
|
# ===== Credential Copy =====
|
|
|
|
async def copy_password(self, cipher_id: str) -> dict:
|
|
"""Get password for clipboard copy."""
|
|
password = self._get_cipher_field(cipher_id, "password")
|
|
if password is None:
|
|
return {"success": False, "error": "Password not found"}
|
|
return {"success": True, "value": password, "clear_after": 60}
|
|
|
|
async def copy_username(self, cipher_id: str) -> dict:
|
|
"""Get username for clipboard copy."""
|
|
username = self._get_cipher_field(cipher_id, "username")
|
|
if username is None:
|
|
return {"success": False, "error": "Username not found"}
|
|
return {"success": True, "value": username, "clear_after": 60}
|
|
|
|
async def copy_totp(self, cipher_id: str) -> dict:
|
|
"""Get TOTP code for clipboard copy."""
|
|
from totp import generate_totp, get_totp_remaining_seconds
|
|
totp_secret = self._get_cipher_field(cipher_id, "totp")
|
|
if totp_secret:
|
|
code = generate_totp(totp_secret)
|
|
remaining = get_totp_remaining_seconds()
|
|
return {"success": True, "value": code, "clear_after": min(remaining, 30), "remaining_seconds": remaining}
|
|
return {"success": False, "error": "TOTP not configured"}
|
|
|
|
async def get_totp_code(self, cipher_id: str) -> dict:
|
|
"""Get current TOTP code without copying."""
|
|
from totp import generate_totp, get_totp_remaining_seconds
|
|
totp_secret = self._get_cipher_field(cipher_id, "totp")
|
|
if totp_secret:
|
|
code = generate_totp(totp_secret)
|
|
remaining = get_totp_remaining_seconds()
|
|
return {"success": True, "code": code, "remaining_seconds": remaining}
|
|
return {"success": False, "error": "TOTP not configured"}
|
|
|
|
# ===== PIN Management =====
|
|
|
|
async def setup_pin(self, pin: str) -> dict:
|
|
"""Set up PIN for vault unlock."""
|
|
try:
|
|
if not self._enc_key:
|
|
return {"success": False, "error": "Vault not unlocked. Login first."}
|
|
|
|
settings = self._load_settings()
|
|
kdf_info = settings.get("kdf_info", {})
|
|
email = settings.get("email", "")
|
|
|
|
# Generate salt for PIN key derivation
|
|
pin_salt = f"{email}:{pin}".lower()
|
|
|
|
# Derive PIN key
|
|
pin_key = self.pin_crypto.derive_pin_key(
|
|
pin, pin_salt, kdf_iterations=200000
|
|
)
|
|
|
|
# Encrypt user key with PIN key
|
|
envelope, iv = self.pin_crypto.encrypt_user_key_for_pin(
|
|
self._enc_key, pin_key
|
|
)
|
|
|
|
# Save PIN settings
|
|
settings["pin_enabled"] = True
|
|
settings["pin_salt"] = pin_salt
|
|
settings["pin_kdf_iterations"] = 200000
|
|
settings["pin_envelope"] = envelope.hex()
|
|
settings["pin_envelope_iv"] = iv.hex()
|
|
self._save_settings(settings)
|
|
|
|
self._pin_attempts = 0
|
|
return {"success": True}
|
|
|
|
except Exception as e:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
async def unlock_with_pin(self, pin: str) -> dict:
|
|
"""Unlock vault using PIN."""
|
|
try:
|
|
settings = self._load_settings()
|
|
|
|
if not settings.get("pin_enabled"):
|
|
return {"success": False, "error": "PIN not configured"}
|
|
|
|
if self._pin_attempts >= self._MAX_PIN_ATTEMPTS:
|
|
return {"success": False, "error": "Too many failed attempts. Login with master password."}
|
|
|
|
pin_salt = settings.get("pin_salt", "")
|
|
pin_iterations = settings.get("pin_kdf_iterations", 200000)
|
|
envelope_hex = settings.get("pin_envelope", "")
|
|
|
|
if not envelope_hex or not pin_salt:
|
|
return {"success": False, "error": "Invalid PIN configuration"}
|
|
|
|
# Derive PIN key
|
|
pin_key = self.pin_crypto.derive_pin_key(
|
|
pin, pin_salt, kdf_iterations=pin_iterations
|
|
)
|
|
|
|
# Decrypt user key from envelope
|
|
envelope = bytes.fromhex(envelope_hex)
|
|
user_key = self.pin_crypto.decrypt_user_key_with_pin(envelope, pin_key)
|
|
|
|
if len(user_key) == 64:
|
|
self._enc_key = user_key[:32]
|
|
self._mac_key = user_key[32:]
|
|
elif len(user_key) == 32:
|
|
self._enc_key = user_key
|
|
self._mac_key = user_key
|
|
else:
|
|
raise ValueError("Invalid decrypted user key length")
|
|
|
|
# Re-sync and decrypt vault
|
|
if self.client:
|
|
sync_data = await self.client.sync_vault()
|
|
self._vault_data = sync_data
|
|
self._decrypted_vault = self.client.decrypt_vault(
|
|
sync_data, self._enc_key, self._mac_key
|
|
)
|
|
self._session_active = True
|
|
self._pin_attempts = 0
|
|
return {"success": True, "vault_size": len(self._decrypted_vault.get("ciphers", []))}
|
|
|
|
return {"success": False, "error": "No active session"}
|
|
|
|
except Exception as e:
|
|
self._pin_attempts += 1
|
|
remaining = self._MAX_PIN_ATTEMPTS - self._pin_attempts
|
|
return {
|
|
"success": False,
|
|
"error": f"Invalid PIN. {remaining} attempts remaining.",
|
|
"attempts_remaining": remaining,
|
|
}
|
|
|
|
async def remove_pin(self) -> dict:
|
|
"""Remove PIN configuration."""
|
|
settings = self._load_settings()
|
|
settings["pin_enabled"] = False
|
|
settings.pop("pin_salt", None)
|
|
settings.pop("pin_kdf_iterations", None)
|
|
settings.pop("pin_envelope", None)
|
|
settings.pop("pin_envelope_iv", None)
|
|
self._save_settings(settings)
|
|
return {"success": True}
|
|
|
|
async def is_pin_enabled(self) -> dict:
|
|
"""Check if PIN is configured."""
|
|
settings = self._load_settings()
|
|
return {"enabled": settings.get("pin_enabled", False)}
|
|
|
|
# ===== Vault Lock =====
|
|
|
|
async def lock_vault(self) -> dict:
|
|
"""Lock the vault and clear sensitive data."""
|
|
from clipboard import clear_all_clipboards
|
|
await clear_all_clipboards()
|
|
self._decrypted_vault = None
|
|
self._vault_data = None
|
|
self._enc_key = None
|
|
self._mac_key = None
|
|
self._session_active = False
|
|
self._pin_attempts = 0
|
|
return {"success": True}
|
|
|
|
async def logout(self) -> dict:
|
|
"""Full logout - clear all data."""
|
|
await self.lock_vault()
|
|
self._master_key = None
|
|
self.client = None
|
|
return {"success": True}
|
|
|
|
# ===== Settings =====
|
|
|
|
async def get_settings(self) -> dict:
|
|
"""Get plugin settings."""
|
|
return {"success": True, "settings": self._load_settings()}
|
|
|
|
async def update_settings(self, new_settings: dict) -> dict:
|
|
"""Update plugin settings."""
|
|
settings = self._load_settings()
|
|
settings.update(new_settings)
|
|
self._save_settings(settings)
|
|
return {"success": True}
|
|
|
|
async def get_status(self) -> dict:
|
|
"""Get current plugin status."""
|
|
return {
|
|
"session_active": self._session_active,
|
|
"vault_loaded": self._decrypted_vault is not None,
|
|
"cipher_count": len(self._decrypted_vault.get("ciphers", [])) if self._decrypted_vault else 0,
|
|
"folder_count": len(self._decrypted_vault.get("folders", [])) if self._decrypted_vault else 0,
|
|
}
|
|
|
|
# ===== Internal Helpers =====
|
|
|
|
def _get_cipher_field(self, cipher_id: str, field: str):
|
|
"""Get a field from a decrypted cipher."""
|
|
if not self._decrypted_vault:
|
|
return None
|
|
|
|
for cipher in self._decrypted_vault.get("ciphers", []):
|
|
if cipher.get("id") == cipher_id:
|
|
if field == "password":
|
|
return cipher.get("login", {}).get("password")
|
|
elif field == "username":
|
|
return cipher.get("login", {}).get("username")
|
|
elif field == "totp":
|
|
return cipher.get("login", {}).get("totp")
|
|
elif field == "name":
|
|
return cipher.get("name")
|
|
return None
|
|
return None
|
|
|
|
def _load_settings(self) -> dict:
|
|
"""Load settings from disk."""
|
|
settings_file = os.path.join(self._settings_path, "settings.json")
|
|
try:
|
|
with open(settings_file, "r") as f:
|
|
return json.load(f)
|
|
except (FileNotFoundError, json.JSONDecodeError):
|
|
return {}
|
|
|
|
def _save_settings(self, settings: dict):
|
|
"""Save settings to disk."""
|
|
os.makedirs(self._settings_path, exist_ok=True)
|
|
settings_file = os.path.join(self._settings_path, "settings.json")
|
|
with open(settings_file, "w") as f:
|
|
json.dump(settings, f, indent=2)
|