294 lines
10 KiB
Python
Executable File
294 lines
10 KiB
Python
Executable File
#!/usr/bin/python3
|
|
# This file is a part of gnome-pass-search-provider.
|
|
#
|
|
# gnome-pass-search-provider is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# gnome-pass-search-provider is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with gnome-pass-search-provider. If not, see
|
|
# <http://www.gnu.org/licenses/>.
|
|
# Copyright (C) 2017 Jonathan Lestrelin
|
|
# Author: Jonathan Lestrelin <jonathan.lestrelin@gmail.com>
|
|
# This project was based on gnome-shell-search-github-repositories
|
|
# Copyright (C) 2012 Red Hat, Inc.
|
|
# Author: Ralph Bean <rbean@redhat.com>
|
|
# which itself was based on fedmsg-notify
|
|
# Copyright (C) 2012 Red Hat, Inc.
|
|
# Author: Luke Macken <lmacken@redhat.com>
|
|
import re
|
|
import subprocess
|
|
from os import getenv
|
|
from os import walk
|
|
from os.path import expanduser
|
|
from os.path import join as path_join
|
|
|
|
import dbus.service
|
|
from dbus.mainloop.glib import DBusGMainLoop
|
|
|
|
try:
|
|
from thefuzz import fuzz
|
|
from thefuzz import process
|
|
except ModuleNotFoundError:
|
|
from fuzzywuzzy import fuzz
|
|
from fuzzywuzzy import process
|
|
from gi.repository import GLib
|
|
|
|
# Convenience shorthand for declaring dbus interface methods.
|
|
# s.b.n. -> search_bus_name
|
|
search_bus_name = "org.gnome.Shell.SearchProvider2"
|
|
sbn = dict(dbus_interface=search_bus_name)
|
|
|
|
|
|
class SearchPassService(dbus.service.Object):
|
|
"""The pass search daemon.
|
|
|
|
This service is started through DBus activation by calling the
|
|
:meth:`Enable` method, and stopped with :meth:`Disable`.
|
|
|
|
"""
|
|
|
|
bus_name = "org.gnome.Pass.SearchProvider"
|
|
_object_path = "/" + bus_name.replace(".", "/")
|
|
|
|
def __init__(self):
|
|
self.session_bus = dbus.SessionBus()
|
|
bus_name = dbus.service.BusName(self.bus_name, bus=self.session_bus)
|
|
dbus.service.Object.__init__(self, bus_name, self._object_path)
|
|
self.password_store = getenv("PASSWORD_STORE_DIR") or expanduser(
|
|
"~/.password-store"
|
|
)
|
|
self.password_executable = getenv("PASSWORD_EXECUTABLE") or "pass"
|
|
self.password_mode = getenv("PASSWORD_MODE") or "pass"
|
|
self.clipboard_executable = getenv("CLIPBOARD_EXECUTABLE") or "wl-copy"
|
|
self.disable_notifications = (
|
|
getenv("DISABLE_NOTIFICATIONS", "false").lower() == "true"
|
|
)
|
|
|
|
@dbus.service.method(in_signature="sasu", **sbn)
|
|
def ActivateResult(self, id, terms, timestamp):
|
|
self.send_password_to_clipboard(id)
|
|
|
|
@dbus.service.method(in_signature="as", out_signature="as", **sbn)
|
|
def GetInitialResultSet(self, terms):
|
|
if self.password_mode == "bw":
|
|
return self.get_bw_result_set(terms)
|
|
else:
|
|
return self.get_pass_result_set(terms)
|
|
|
|
@dbus.service.method(in_signature="as", out_signature="aa{sv}", **sbn)
|
|
def GetResultMetas(self, ids):
|
|
return [
|
|
dict(
|
|
id=id,
|
|
name=id[1:] if id.startswith(":") else id,
|
|
gicon="dialog-password",
|
|
)
|
|
for id in ids
|
|
]
|
|
|
|
@dbus.service.method(in_signature="asas", out_signature="as", **sbn)
|
|
def GetSubsearchResultSet(self, previous_results, new_terms):
|
|
if self.password_mode == "bw":
|
|
return self.get_bw_result_set(new_terms)
|
|
else:
|
|
return self.get_pass_result_set(new_terms)
|
|
|
|
@dbus.service.method(in_signature="asu", terms="as", timestamp="u", **sbn)
|
|
def LaunchSearch(self, terms, timestamp):
|
|
pass
|
|
|
|
def get_bw_result_set(self, terms):
|
|
if terms[0].startswith(":"):
|
|
field = terms[0][1:]
|
|
terms = terms[1:]
|
|
else:
|
|
field = None
|
|
name = "".join(terms)
|
|
|
|
password_list = subprocess.check_output(
|
|
[self.password_executable, "list"], universal_newlines=True
|
|
).split("\n")[:-1]
|
|
|
|
results = list(filter(lambda p: p.startswith(name), password_list))[:5]
|
|
|
|
remaining = 5 - len(results)
|
|
|
|
if remaining > 0:
|
|
fuzzy_results = [
|
|
e[0]
|
|
for e in process.extract(
|
|
name, password_list, limit=remaining, scorer=fuzz.token_sort_ratio
|
|
)
|
|
]
|
|
|
|
results.extend(fuzzy_results)
|
|
|
|
if field is not None:
|
|
results = [f":{field} {r}" for r in results]
|
|
return results
|
|
|
|
def get_pass_result_set(self, terms):
|
|
if terms[0] == "otp":
|
|
field = terms[0]
|
|
elif terms[0].startswith(":"):
|
|
field = terms[0][1:]
|
|
terms = terms[1:]
|
|
else:
|
|
field = None
|
|
|
|
name = "".join(terms)
|
|
password_list = []
|
|
for root, dirs, files in walk(self.password_store):
|
|
dir_path = root[len(self.password_store) + 1 :]
|
|
|
|
if dir_path.startswith("."):
|
|
continue
|
|
|
|
for filename in files:
|
|
if filename[-4:] != ".gpg":
|
|
continue
|
|
path = path_join(dir_path, filename)[:-4]
|
|
password_list.append(path)
|
|
|
|
results = list(sorted(filter(lambda p: p.startswith(name), password_list)))[:5]
|
|
|
|
remaining = 5 - len(results)
|
|
|
|
if remaining > 0:
|
|
containing_results = list(
|
|
sorted(filter(lambda p: name in p, password_list))
|
|
)[:remaining]
|
|
|
|
results.extend(containing_results)
|
|
|
|
remaining = 5 - len(results)
|
|
|
|
if remaining > 0:
|
|
fuzzy_results = [
|
|
e[0]
|
|
for e in process.extract(
|
|
name, password_list, limit=remaining, scorer=fuzz.token_sort_ratio
|
|
)
|
|
]
|
|
|
|
results.extend(fuzzy_results)
|
|
|
|
if field == "otp":
|
|
results = [f"otp {r}" for r in results]
|
|
elif field is not None:
|
|
results = [f":{field} {r}" for r in results]
|
|
return results
|
|
|
|
def get_value_from_output(self, output, field=None):
|
|
if field is not None:
|
|
match = re.search(
|
|
rf"^{field}:\s*(?P<value>.+?)$", output, flags=re.I | re.M
|
|
)
|
|
if match:
|
|
value = match.group("value")
|
|
else:
|
|
raise RuntimeError(
|
|
f"The field {field} was not found in the password entry."
|
|
)
|
|
else:
|
|
value = output.split("\n", 1)[0]
|
|
return value
|
|
|
|
def send_password_to_gpaste(self, base_args, name, field=None):
|
|
gpaste = self.session_bus.get_object(
|
|
"org.gnome.GPaste.Daemon", "/org/gnome/GPaste"
|
|
)
|
|
|
|
output = subprocess.check_output(base_args + [name], universal_newlines=True)
|
|
value = self.get_value_from_output(output, field)
|
|
try:
|
|
gpaste.AddPassword(name, value, dbus_interface="org.gnome.GPaste1")
|
|
except dbus.DBusException:
|
|
gpaste.AddPassword(name, value, dbus_interface="org.gnome.GPaste2")
|
|
|
|
def send_password_to_native_clipboard(self, base_args, name, field=None):
|
|
if self.password_mode == "bw":
|
|
output = subprocess.check_output(
|
|
base_args + [name], universal_newlines=True
|
|
)
|
|
value = self.get_value_from_output(output, field)
|
|
p1 = subprocess.run(self.clipboard_executable, input=value, text=True)
|
|
if p1.returncode:
|
|
raise RuntimeError(
|
|
f"Error while running copying to clipboard: got return code: {p1.returncode}."
|
|
)
|
|
else:
|
|
if field is not None:
|
|
raise RuntimeError("This feature requires GPaste.")
|
|
result = subprocess.run(base_args + ["-c", name])
|
|
if result.returncode:
|
|
raise RuntimeError(
|
|
f"Error while running pass: got return code: {result.returncode}."
|
|
)
|
|
|
|
def send_password_to_clipboard(self, name):
|
|
if name.startswith(":"):
|
|
field, name = name.split(" ", 1)
|
|
field = field[1:]
|
|
else:
|
|
field = None
|
|
|
|
if self.password_mode == "bw":
|
|
base_args = [self.password_executable, "get", "--full"]
|
|
elif name.startswith("otp "):
|
|
base_args = [self.password_executable, "otp", "code"]
|
|
name = name[4:]
|
|
field = None
|
|
else:
|
|
base_args = [self.password_executable, "show"]
|
|
|
|
try:
|
|
try:
|
|
self.send_password_to_gpaste(base_args, name, field)
|
|
except dbus.DBusException:
|
|
# We couldn't join GPaste over D-Bus, use native clipboard
|
|
self.send_password_to_native_clipboard(base_args, name, field)
|
|
if "otp" in base_args:
|
|
self.notify("Copied OTP password to clipboard:", body=f"<b>{name}</b>")
|
|
elif field is not None:
|
|
self.notify(
|
|
f"Copied field {field} to clipboard:", body=f"<b>{name}</b>"
|
|
)
|
|
else:
|
|
self.notify("Copied password to clipboard:", body=f"<b>{name}</b>")
|
|
except (subprocess.CalledProcessError, FileNotFoundError, RuntimeError) as e:
|
|
self.notify("Failed to copy password or field!", body=str(e), error=True)
|
|
|
|
def notify(self, message, body="", error=False):
|
|
if not error and self.disable_notifications:
|
|
return
|
|
try:
|
|
self.session_bus.get_object(
|
|
"org.freedesktop.Notifications", "/org/freedesktop/Notifications"
|
|
).Notify(
|
|
"Pass",
|
|
0,
|
|
"dialog-password",
|
|
message,
|
|
body,
|
|
"",
|
|
{"transient": False if error else True},
|
|
0 if error else 3000,
|
|
dbus_interface="org.freedesktop.Notifications",
|
|
)
|
|
except dbus.DBusException as err:
|
|
print(f"Error {err} while trying to display {message}.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
DBusGMainLoop(set_as_default=True)
|
|
SearchPassService()
|
|
GLib.MainLoop().run()
|