""" 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 sync_vault(self) -> dict: """Sync vault from server and re-decrypt.""" try: if not self.client or not self._session_active: return {"success": False, "error": "No active session"} if not self._enc_key or not self._mac_key: return {"success": False, "error": "Encryption keys not available"} 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, "vault_size": len(self._decrypted_vault.get("ciphers", []))} except Exception as e: return {"success": False, "error": str(e)} 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 (store both enc_key + mac_key) envelope, iv = self.pin_crypto.encrypt_user_key_for_pin( self._enc_key + self._mac_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() # Persist tokens encrypted with PIN for session restore across restarts if self.client and self.client.access_token: import json as _json token_data = _json.dumps({ "access_token": self.client.access_token, "refresh_token": self.client.refresh_token or "", "user_id": self.client.user_id or "", }).encode("utf-8") token_envelope, token_iv = self.pin_crypto.encrypt_user_key_for_pin( token_data, pin_key ) settings["pin_token_envelope"] = token_envelope.hex() settings["pin_token_envelope_iv"] = token_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: # Old format: only enc_key was stored, mac_key is wrong. # Clear stale PIN and force re-setup. 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) settings.pop("pin_token_envelope", None) settings.pop("pin_token_envelope_iv", None) self._save_settings(settings) return {"success": False, "error": "PIN was set with an older version. Please log in and set up PIN again."} else: raise ValueError("Invalid decrypted user key length") # Restore session from persisted tokens if no active client (restart scenario) if not self.client: token_envelope_hex = settings.get("pin_token_envelope", "") if token_envelope_hex: import json as _json token_envelope = bytes.fromhex(token_envelope_hex) token_data_bytes = self.pin_crypto.decrypt_user_key_with_pin( token_envelope, pin_key ) token_data = _json.loads(token_data_bytes.decode("utf-8")) server_url = settings.get("server_url", "https://api.bitwarden.com") self.client = self.client_class(server_url) self.client.access_token = token_data.get("access_token", "") self.client.refresh_token = token_data.get("refresh_token", "") self.client.user_id = token_data.get("user_id", "") else: return {"success": False, "error": "No active session and no persisted tokens"} # Re-sync and decrypt vault 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", []))} 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) settings.pop("pin_token_envelope", None) settings.pop("pin_token_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)} async def get_saved_credentials(self) -> dict: """Get saved server URL and email for pre-filling login form.""" settings = self._load_settings() return { "success": True, "server_url": settings.get("server_url", "https://api.bitwarden.com"), "email": settings.get("email", ""), "pin_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 # Clear persisted tokens from settings settings = self._load_settings() settings.pop("pin_token_envelope", None) settings.pop("pin_token_envelope_iv", None) self._save_settings(settings) 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)