Linting with black.
This commit is contained in:
parent
2ffc11ce05
commit
da2db4130e
@ -38,7 +38,7 @@ from os.path import expanduser, join as path_join
|
|||||||
|
|
||||||
# Convenience shorthand for declaring dbus interface methods.
|
# Convenience shorthand for declaring dbus interface methods.
|
||||||
# s.b.n. -> search_bus_name
|
# s.b.n. -> search_bus_name
|
||||||
search_bus_name = 'org.gnome.Shell.SearchProvider2'
|
search_bus_name = "org.gnome.Shell.SearchProvider2"
|
||||||
sbn = dict(dbus_interface=search_bus_name)
|
sbn = dict(dbus_interface=search_bus_name)
|
||||||
|
|
||||||
|
|
||||||
@ -49,128 +49,143 @@ class SearchPassService(dbus.service.Object):
|
|||||||
:meth:`Enable` method, and stopped with :meth:`Disable`.
|
:meth:`Enable` method, and stopped with :meth:`Disable`.
|
||||||
|
|
||||||
"""
|
"""
|
||||||
bus_name = 'org.gnome.Pass.SearchProvider'
|
|
||||||
_object_path = '/' + bus_name.replace('.', '/')
|
bus_name = "org.gnome.Pass.SearchProvider"
|
||||||
|
_object_path = "/" + bus_name.replace(".", "/")
|
||||||
|
|
||||||
def __init__(self):
|
def __init__(self):
|
||||||
self.session_bus = dbus.SessionBus()
|
self.session_bus = dbus.SessionBus()
|
||||||
bus_name = dbus.service.BusName(self.bus_name, bus=self.session_bus)
|
bus_name = dbus.service.BusName(self.bus_name, bus=self.session_bus)
|
||||||
dbus.service.Object.__init__(self, bus_name, self._object_path)
|
dbus.service.Object.__init__(self, bus_name, self._object_path)
|
||||||
self.password_store = getenv('PASSWORD_STORE_DIR') or \
|
self.password_store = getenv("PASSWORD_STORE_DIR") or expanduser(
|
||||||
expanduser('~/.password-store')
|
"~/.password-store"
|
||||||
|
)
|
||||||
|
|
||||||
@dbus.service.method(in_signature='sasu', **sbn)
|
@dbus.service.method(in_signature="sasu", **sbn)
|
||||||
def ActivateResult(self, id, terms, timestamp):
|
def ActivateResult(self, id, terms, timestamp):
|
||||||
self.send_password_to_clipboard(id)
|
self.send_password_to_clipboard(id)
|
||||||
|
|
||||||
@dbus.service.method(in_signature='as', out_signature='as', **sbn)
|
@dbus.service.method(in_signature="as", out_signature="as", **sbn)
|
||||||
def GetInitialResultSet(self, terms):
|
def GetInitialResultSet(self, terms):
|
||||||
return self.get_result_set(terms)
|
return self.get_result_set(terms)
|
||||||
|
|
||||||
@dbus.service.method(in_signature='as', out_signature='aa{sv}', **sbn)
|
@dbus.service.method(in_signature="as", out_signature="aa{sv}", **sbn)
|
||||||
def GetResultMetas(self, ids):
|
def GetResultMetas(self, ids):
|
||||||
return [dict(id=id, name=id[1:] if id.startswith(':') else id,
|
return [
|
||||||
gicon='dialog-password') for id in ids]
|
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)
|
@dbus.service.method(in_signature="asas", out_signature="as", **sbn)
|
||||||
def GetSubsearchResultSet(self, previous_results, new_terms):
|
def GetSubsearchResultSet(self, previous_results, new_terms):
|
||||||
return self.get_result_set(new_terms)
|
return self.get_result_set(new_terms)
|
||||||
|
|
||||||
@dbus.service.method(in_signature='asu', terms='as', timestamp='u', **sbn)
|
@dbus.service.method(in_signature="asu", terms="as", timestamp="u", **sbn)
|
||||||
def LaunchSearch(self, terms, timestamp):
|
def LaunchSearch(self, terms, timestamp):
|
||||||
pass
|
pass
|
||||||
|
|
||||||
def get_result_set(self, terms):
|
def get_result_set(self, terms):
|
||||||
if terms[0] == 'otp':
|
if terms[0] == "otp":
|
||||||
field = terms[0]
|
field = terms[0]
|
||||||
elif terms[0].startswith(':'):
|
elif terms[0].startswith(":"):
|
||||||
field = terms[0][1:]
|
field = terms[0][1:]
|
||||||
terms = terms[1:]
|
terms = terms[1:]
|
||||||
else:
|
else:
|
||||||
field = None
|
field = None
|
||||||
|
|
||||||
name = ''.join(terms)
|
name = "".join(terms)
|
||||||
password_list = []
|
password_list = []
|
||||||
for root, dirs, files in walk(self.password_store):
|
for root, dirs, files in walk(self.password_store):
|
||||||
dir_path = root[len(self.password_store) + 1:]
|
dir_path = root[len(self.password_store) + 1 :]
|
||||||
|
|
||||||
if dir_path.startswith('.'):
|
if dir_path.startswith("."):
|
||||||
continue
|
continue
|
||||||
|
|
||||||
for filename in files:
|
for filename in files:
|
||||||
if filename[-4:] != '.gpg':
|
if filename[-4:] != ".gpg":
|
||||||
continue
|
continue
|
||||||
path = path_join(dir_path, filename)[:-4]
|
path = path_join(dir_path, filename)[:-4]
|
||||||
password_list.append(path)
|
password_list.append(path)
|
||||||
|
|
||||||
results = [e[0] for e in process.extract(name, password_list, limit=5,
|
results = [
|
||||||
scorer=fuzz.partial_ratio)]
|
e[0]
|
||||||
if field == 'otp':
|
for e in process.extract(
|
||||||
results = [f'otp {r}' for r in results]
|
name, password_list, limit=5, scorer=fuzz.partial_ratio
|
||||||
|
)
|
||||||
|
]
|
||||||
|
if field == "otp":
|
||||||
|
results = [f"otp {r}" for r in results]
|
||||||
elif field is not None:
|
elif field is not None:
|
||||||
results = [f':{field} {r}' for r in results]
|
results = [f":{field} {r}" for r in results]
|
||||||
return results
|
return results
|
||||||
|
|
||||||
def send_password_to_gpaste(self, base_args, name, field=None):
|
def send_password_to_gpaste(self, base_args, name, field=None):
|
||||||
try:
|
try:
|
||||||
gpaste = self.session_bus.get_object('org.gnome.GPaste.Daemon',
|
gpaste = self.session_bus.get_object(
|
||||||
'/org/gnome/GPaste')
|
"org.gnome.GPaste.Daemon", "/org/gnome/GPaste"
|
||||||
|
)
|
||||||
|
|
||||||
output = subprocess.check_output(base_args + [name],
|
output = subprocess.check_output(
|
||||||
stderr=subprocess.STDOUT,
|
base_args + [name], stderr=subprocess.STDOUT, universal_newlines=True
|
||||||
universal_newlines=True)
|
)
|
||||||
if field is not None:
|
if field is not None:
|
||||||
match = re.search(fr'^{field}:\s*(?P<value>.+?)$', output,
|
match = re.search(
|
||||||
flags=re.I | re.M)
|
fr"^{field}:\s*(?P<value>.+?)$", output, flags=re.I | re.M
|
||||||
|
)
|
||||||
if match:
|
if match:
|
||||||
password = match.group('value')
|
password = match.group("value")
|
||||||
else:
|
else:
|
||||||
raise RuntimeError(f'The field {field} was not found in ' +
|
raise RuntimeError(
|
||||||
'the pass file.')
|
f"The field {field} was not found in " + "the pass file."
|
||||||
|
)
|
||||||
else:
|
else:
|
||||||
password = output.split('\n', 1)[0]
|
password = output.split("\n", 1)[0]
|
||||||
|
|
||||||
gpaste.AddPassword(name, password,
|
gpaste.AddPassword(name, password, dbus_interface="org.gnome.GPaste1")
|
||||||
dbus_interface='org.gnome.GPaste1')
|
|
||||||
|
|
||||||
if 'otp' in base_args:
|
if "otp" in base_args:
|
||||||
self.notify('Copied OTP password to clipboard:',
|
self.notify("Copied OTP password to clipboard:", body=f"<b>{name}</b>")
|
||||||
body=f'<b>{name}</b>')
|
|
||||||
elif field is not None:
|
elif field is not None:
|
||||||
self.notify(f'Copied field {field} to clipboard:',
|
self.notify(
|
||||||
body=f'<b>{name}</b>')
|
f"Copied field {field} to clipboard:", body=f"<b>{name}</b>"
|
||||||
|
)
|
||||||
else:
|
else:
|
||||||
self.notify('Copied password to clipboard:',
|
self.notify("Copied password to clipboard:", body=f"<b>{name}</b>")
|
||||||
body=f'<b>{name}</b>')
|
|
||||||
except subprocess.CalledProcessError as e:
|
except subprocess.CalledProcessError as e:
|
||||||
self.notify('Failed to copy password!', body=e.output, error=True)
|
self.notify("Failed to copy password!", body=e.output, error=True)
|
||||||
except RuntimeError as e:
|
except RuntimeError as e:
|
||||||
self.notify('Failed to copy field!', body=e.output, error=True)
|
self.notify("Failed to copy field!", body=e.output, error=True)
|
||||||
|
|
||||||
def send_password_to_native_clipboard(self, base_args, name, field=None):
|
def send_password_to_native_clipboard(self, base_args, name, field=None):
|
||||||
if field is not None:
|
if field is not None:
|
||||||
self.notify(f'Cannot copy field values.',
|
self.notify(
|
||||||
body='This feature requires GPaste.', error=True)
|
f"Cannot copy field values.",
|
||||||
|
body="This feature requires GPaste.",
|
||||||
|
error=True,
|
||||||
|
)
|
||||||
return
|
return
|
||||||
|
|
||||||
pass_cmd = subprocess.run(base_args + ['-c', name])
|
pass_cmd = subprocess.run(base_args + ["-c", name])
|
||||||
if pass_cmd.returncode:
|
if pass_cmd.returncode:
|
||||||
self.notify('Failed to copy password!', error=True)
|
self.notify("Failed to copy password!", error=True)
|
||||||
elif 'otp' in base_args:
|
elif "otp" in base_args:
|
||||||
self.notify('Copied OTP password to clipboard:',
|
self.notify("Copied OTP password to clipboard:", body=f"<b>{name}</b>")
|
||||||
body=f'<b>{name}</b>')
|
|
||||||
else:
|
else:
|
||||||
self.notify('Copied password to clipboard:', body=f'<b>{name}</b>')
|
self.notify("Copied password to clipboard:", body=f"<b>{name}</b>")
|
||||||
|
|
||||||
def send_password_to_clipboard(self, name):
|
def send_password_to_clipboard(self, name):
|
||||||
field = None
|
field = None
|
||||||
if name.startswith('otp '):
|
if name.startswith("otp "):
|
||||||
base_args = ['pass', 'otp', 'code']
|
base_args = ["pass", "otp", "code"]
|
||||||
name = name[4:]
|
name = name[4:]
|
||||||
else:
|
else:
|
||||||
base_args = ['pass', 'show']
|
base_args = ["pass", "show"]
|
||||||
if name.startswith(':'):
|
if name.startswith(":"):
|
||||||
field, name = name.split(' ', 1)
|
field, name = name.split(" ", 1)
|
||||||
field = field[1:]
|
field = field[1:]
|
||||||
|
|
||||||
try:
|
try:
|
||||||
@ -180,27 +195,26 @@ class SearchPassService(dbus.service.Object):
|
|||||||
# use pass native clipboard copy
|
# use pass native clipboard copy
|
||||||
self.send_password_to_native_clipboard(base_args, name, field)
|
self.send_password_to_native_clipboard(base_args, name, field)
|
||||||
|
|
||||||
def notify(self, message, body='', error=False):
|
def notify(self, message, body="", error=False):
|
||||||
try:
|
try:
|
||||||
self.session_bus.get_object(
|
self.session_bus.get_object(
|
||||||
'org.freedesktop.Notifications',
|
"org.freedesktop.Notifications", "/org/freedesktop/Notifications"
|
||||||
'/org/freedesktop/Notifications'
|
|
||||||
).Notify(
|
).Notify(
|
||||||
'Pass',
|
"Pass",
|
||||||
0,
|
0,
|
||||||
'dialog-password',
|
"dialog-password",
|
||||||
message,
|
message,
|
||||||
body,
|
body,
|
||||||
'',
|
"",
|
||||||
{'transient': False if error else True},
|
{"transient": False if error else True},
|
||||||
0 if error else 3000,
|
0 if error else 3000,
|
||||||
dbus_interface='org.freedesktop.Notifications'
|
dbus_interface="org.freedesktop.Notifications",
|
||||||
)
|
)
|
||||||
except dbus.DBusException as err:
|
except dbus.DBusException as err:
|
||||||
print(f'Error {err} while trying to display {message}.')
|
print(f"Error {err} while trying to display {message}.")
|
||||||
|
|
||||||
|
|
||||||
if __name__ == '__main__':
|
if __name__ == "__main__":
|
||||||
DBusGMainLoop(set_as_default=True)
|
DBusGMainLoop(set_as_default=True)
|
||||||
SearchPassService()
|
SearchPassService()
|
||||||
GLib.MainLoop().run()
|
GLib.MainLoop().run()
|
||||||
|
Loading…
x
Reference in New Issue
Block a user