initial commit (wip)

3 files changed, 99 insertions, 0 deletions

diff --git a/nmpass/__init__.py b/nmpass/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/nmpass/__init__.py
diff --git a/nmpass/main.py b/nmpass/main.py
new file mode 100644
index 0000000..e969e13
--- /dev/null
+++ b/nmpass/main.py
@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+from asyncio import new_event_loop
+from sdbus_async.networkmanager import NetworkManagerSecretAgentInterfaceAsync, NetworkManagerAgentManager, NetworkManagerConnectionProperties
+from sdbus_async.networkmanager.settings import ConnectionProfile
+from typing import Any
+import sdbus
+
+from .store import PasswordStore
+
+class NetworkManagerPasswordStoreAgent(NetworkManagerSecretAgentInterfaceAsync):
+  store = PasswordStore()
+
+  @sdbus.dbus_method_async_override()
+  async def get_secrets(
+    self,
+    connection: NetworkManagerConnectionProperties,
+    _connection_path: str,
+    setting_name: str,
+    _hints: list[str],
+    _flags: int,
+  ) -> dict[str, dict[str, tuple[str, Any]]]:
+    profile = ConnectionProfile.from_dbus(connection)
+
+    printf(setting_name)
+    if setting_name == '802-11-wireless-security':
+      ssid = profile.wireless.ssid.decode()
+      pass_name = f"net/{ssid}/passwd"
+      password = self.store.retrieve(pass_name)
+      printf(f"SSID: {ssid} PASSWORD: {password}")
+      if password != None:
+        profile.wireless_security.psk = password
+        return profile.wireless_security.psk.to_dbus()
+
+    return {}
+
+  @sdbus.dbus_method_async_override()
+  async def save_secrets(
+    self,
+    connection: NetworkManagerConnectionProperties,
+    _connection_path: str,
+  ) -> None:
+    profile = ConnectionProfile.from_dbus(connection)
+    print(profile)
+    raise NotImplementedError
+
+def main():
+  sdbus.set_default_bus(sdbus.sd_bus_open_system())
+  agent = NetworkManagerPasswordStoreAgent()
+  loop = new_event_loop()
+  agent.export_to_dbus('/org/freedesktop/NetworkManager/SecretAgent')
+  agent_manager = NetworkManagerAgentManager()
+  try:
+    loop.run_until_complete(agent_manager.register('org.nmpassd'))
+    loop.run_forever()
+  except KeyboardInterrupt:
+    loop.stop()
+
+if __name__ == "__main__":
+  main()
+
diff --git a/nmpass/store.py b/nmpass/store.py
new file mode 100644
index 0000000..08080ba
--- /dev/null
+++ b/nmpass/store.py
@@ -0,0 +1,38 @@
+from subprocess import run, PIPE, DEVNULL
+from os import environ, path
+
+NEWLINE = "\n"
+
+class PasswordStore():
+  PASSWORD_STORE = "pass"
+  PASSWORD_STORE_DIR = path.join(environ["HOME"], ".password-store")
+
+  def __init__(self):
+    if "PASSWORD_STORE_DIR" in environ:
+      self.PASSWORD_STORE_DIR = environ["PASSWORD_STORE_DIR"]
+
+  # There's no way to check using the pass CLI whether a password name passed
+  # to `pass show` is a password or a directory, so this checks manually
+  def exists(self, name: str) -> bool:
+    return path.isfile(path.join(self.PASSWORD_STORE_DIR, f"{name}.gpg"))
+
+  def retrieve(self, name: str, truncate: bool = True) -> str | None:
+    if not self.exists(name):
+      return None
+    cmd = (self.PASSWORD_STORE, 'show', name,)
+    proc = run(cmd, stdout=PIPE)
+    output = proc.stdout.decode()
+    if not truncate:
+      return output
+    return output.split(NEWLINE)[0]
+
+  def store(self, name: str, password: str) -> None:
+    current = self.retrieve(name, False)
+    if current == None:
+      current = NEWLINE
+    to = current.find(NEWLINE)
+    current = f"{password}{current[to:]}"
+
+    cmd = (self.PASSWORD_STORE, 'insert', '--multiline', '--force', name,)
+    run(cmd, input=current.encode(), stdout=DEVNULL, stderr=DEVNULL)
+