""" 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: """Copy a password to clipboard with 60s auto-clear.""" from clipboard import copy_to_clipboard password = self._get_cipher_field(cipher_id, "password") if password is None: return {"success": False, "error": "Password not found"} return await copy_to_clipboard(password, clear_after=60, label=f"password_{cipher_id}") async def copy_username(self, cipher_id: str) -> dict: """Copy a username to clipboard with 60s auto-clear.""" from clipboard import copy_to_clipboard username = self._get_cipher_field(cipher_id, "username") if username is None: return {"success": False, "error": "Username not found"} return await copy_to_clipboard(username, clear_after=60, label=f"username_{cipher_id}") async def copy_totp(self, cipher_id: str) -> dict: """Copy TOTP code to clipboard with 60s auto-clear.""" from clipboard import copy_to_clipboard 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() result = await copy_to_clipboard(code, clear_after=min(remaining, 30), label=f"totp_{cipher_id}") result["remaining_seconds"] = remaining return result 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)