diff --git a/bitwarden/__init__.py b/bitwarden/__init__.py index 6e9e8031..3c8bf20e 100644 --- a/bitwarden/__init__.py +++ b/bitwarden/__init__.py @@ -1,12 +1,15 @@ # -*- coding: utf-8 -*- +import time +from dataclasses import dataclass +from enum import Enum from pathlib import Path -from subprocess import run, CalledProcessError +from subprocess import CalledProcessError, run from albert import * md_iid = "3.0" -md_version = "3.0" +md_version = "3.1" md_name = "Bitwarden" md_description = "'rbw' wrapper extension" md_license = "MIT" @@ -14,8 +17,18 @@ md_authors = ["@ovitor", "@daviddeadly", "@manuelschneid3r"] md_bin_dependencies = ["rbw"] +MAX_MINUTES_CACHE_TIMEOUT = 60 +DEFAULT_MINUTE_CACHE_TIMEOUT = 5 + + +@dataclass(frozen=True) +class ConfigKeys: + CACHE_TIMEOUT = "cache_timeout" + class Plugin(PluginInstance, TriggerQueryHandler): + _cached_items = None + _last_fetch_time = 0 iconUrls = [f"file:{Path(__file__).parent}/bw.svg"] @@ -23,8 +36,36 @@ def __init__(self): PluginInstance.__init__(self) TriggerQueryHandler.__init__(self) + self.cache_timeout = ( + self.readConfig(ConfigKeys.CACHE_TIMEOUT, int) + or DEFAULT_MINUTE_CACHE_TIMEOUT + ) + def defaultTrigger(self): - return 'bw ' + return "bw " + + @property + def cache_timeout(self): + return int(self._cache_timeout / 60) + + @cache_timeout.setter + def cache_timeout(self, value): + self._cache_timeout = int(value * 60) + self.writeConfig(ConfigKeys.CACHE_TIMEOUT, value) + + def configWidget(self): + return [ + { + "type": "label", + "text": "Cache (result of `rbw list`) duration", + }, + { + "type": "spinbox", + "property": ConfigKeys.CACHE_TIMEOUT, + "label": f"Minutes: (max: {MAX_MINUTES_CACHE_TIMEOUT}, disable: 0)", + "widget_properties": {"maximum": MAX_MINUTES_CACHE_TIMEOUT}, + }, + ] def handleTriggerQuery(self, query): if query.string.strip().lower() == "sync": @@ -37,11 +78,9 @@ def handleTriggerQuery(self, query): Action( id="sync", text="Syncing Bitwarden Vault", - callable=lambda: run( - ["rbw", "sync"], - ) + callable=lambda: self._sync_vault(), ) - ] + ], ) ) @@ -56,30 +95,38 @@ def handleTriggerQuery(self, query): Action( id="copy", text="Copy password to clipboard", - callable=lambda item=p: self._password_to_clipboard(item) + callable=lambda item=p: self._password_to_clipboard(item), ), Action( id="copy-auth", text="Copy auth code to clipboard", - callable=lambda item=p: self._code_to_clipboard(item) + callable=lambda item=p: self._code_to_clipboard(item), ), Action( id="copy-username", text="Copy username to clipboard", - callable=lambda username=p["user"]: - setClipboardText(text=username) + callable=lambda username=p["user"]: setClipboardText( + text=username + ), ), Action( id="edit", text="Edit entry in terminal", - callable=lambda item=p: self._edit_entry(item) - ) - ] + callable=lambda item=p: self._edit_entry(item), + ), + ], ) ) - @staticmethod - def _get_items(): + def _get_items(self): + not_first_time = self._cached_items is not None + + time_passed = time.time() - self._last_fetch_time + is_chache_fresh = time_passed < self._cache_timeout + + if not_first_time and is_chache_fresh: + return self._cached_items + field_names = ["id", "name", "user", "folder"] raw_items = run( ["rbw", "list", "--fields", ",".join(field_names)], @@ -98,12 +145,16 @@ def _get_items(): item["path"] = item["folder"] + "/" + item["name"] else: item["path"] = item["name"] + items.append(item) + self._cached_items = items + self._last_fetch_time = time.time() + return items def _filter_items(self, query): - passwords = self._get_items() + passwords = self._get_items() or [] search_fields = ["path", "user"] # Use a set for faster membership tests words = set(query.string.strip().lower().split()) @@ -121,15 +172,18 @@ def _filter_items(self, query): return filtered_passwords + def _sync_vault(self): + run(["rbw", "sync"], check=True) + + self._cached_items = None + self._last_fetch_time = 0 + @staticmethod def _password_to_clipboard(item): rbw_id = item["id"] password = run( - ["rbw", "get", rbw_id], - capture_output=True, - encoding="utf-8", - check=True + ["rbw", "get", rbw_id], capture_output=True, encoding="utf-8", check=True ).stdout.strip() setClipboardText(text=password) @@ -143,14 +197,14 @@ def _code_to_clipboard(item): ["rbw", "code", rbw_id], capture_output=True, encoding="utf-8", - check=True + check=True, ).stdout.strip() except CalledProcessError as err: code = run( ["echo", err.__str__()], capture_output=True, encoding="utf-8", - check=True + check=True, ).stdout.strip() setClipboardText(text=code)