Draw the authentication prompt ourselves

hyprpolkitagent's dialog is compiled into its binary -- no config, no stylesheet, nothing to theme -- and it was the one window on this desktop that looked like it belonged to something else. The split between the two halves is the security design, not an implementation detail. A small agent process owns the D-Bus side: it registers with polkitd, receives the request, and hands the shell the action, the message, who may answer, and a one-time cookie. It never sees a password. The shell draws the prompt and, on submit, spawns the setuid polkit-agent-helper-1 itself and writes the password to that helper's stdin; the helper runs the PAM conversation and reports to polkitd directly. The password exists in the shell and in the helper's stdin and nowhere else -- never on a command line, never over D-Bus, never through IPC arguments. The prompt takes exclusive keyboard focus, because a password field that lets keystrokes reach the window behind it is a keylogger with extra steps. The request travels as a file created 0600 with O_EXCL inside a 0700 runtime directory: a cookie is not a password, but it is a capability, and capabilities do not belong in a process listing either. Three things cost real time. polkitd calls back on the same connection that registered, so exporting the object on the session bus while registering from the system bus failed every request as "Not authorized" with no error anywhere. XDG_SESSION_ID is absent in a systemd user unit, which runs under [email protected] and belongs to no login session, so the session comes from logind's Display property instead. And PyGObject does not accept the @ placeholder in variant format strings. hyprpolkitagent stays installed as the fallback, only one agent is started, and the comment beside the autostart says how to get the stock prompt back. Verified end to end, including a real password accepted and three cancellations refused. Claude-Session: https://claude.ai/code/session_01BRvzt4H8XXLPVH5MyYdk9L
2026-08-19 18:08:52 -04:00
parent b10f8e2593
commit 116510caa8
9 changed files with 867 additions and 1 deletions
@@ -0,0 +1,319 @@
+#!/usr/bin/env python3
+
+"""Panama's polkit authentication agent -- the D-Bus half.
+
+This process never sees a password. It is worth being explicit about why the
+work is split this way, because the split IS the security design:
+
+  * This half registers with polkitd, receives an authentication request, and
+    hands the non-secret parts of it to the shell: which action, what message,
+    which identities may answer, and the one-time cookie.
+  * The shell half draws the prompt, and when someone types a password it
+    spawns the setuid `polkit-agent-helper-1` itself and writes the password to
+    that helper's STDIN. The helper performs the PAM conversation and reports
+    the result to polkitd directly.
+
+So the password exists only inside the shell process and the helper's stdin. It
+never crosses D-Bus, never crosses this process, and never appears on a command
+line -- argv is world-readable through /proc, which is why "just pass it as an
+argument" is not an option anywhere in this codebase.
+
+The request itself is handed over as a file in the runtime directory, created
+0600, rather than as IPC arguments. The cookie is not a password, but it is a
+capability, and capabilities do not belong in a process listing either.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import signal
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+import gi
+
+gi.require_version("Gio", "2.0")
+gi.require_version("GLib", "2.0")
+from gi.repository import Gio, GLib  # noqa: E402
+
+AGENT_PATH = "/org/panama/PolkitAgent"
+AGENT_INTERFACE = "org.freedesktop.PolicyKit1.AuthenticationAgent"
+AUTHORITY_NAME = "org.freedesktop.PolicyKit1"
+AUTHORITY_PATH = "/org/freedesktop/PolicyKit1/Authority"
+AUTHORITY_INTERFACE = "org.freedesktop.PolicyKit1.Authority"
+
+# How long a prompt may stay on screen before this gives up and tells polkit the
+# request failed. Without it, a shell that died mid-prompt would leave the
+# caller (a package manager, say) waiting forever.
+PROMPT_TIMEOUT_SECONDS = 300
+
+INTROSPECTION = """
+<node>
+  <interface name='org.freedesktop.PolicyKit1.AuthenticationAgent'>
+    <method name='BeginAuthentication'>
+      <arg type='s' name='action_id' direction='in'/>
+      <arg type='s' name='message' direction='in'/>
+      <arg type='s' name='icon_name' direction='in'/>
+      <arg type='a{ss}' name='details' direction='in'/>
+      <arg type='s' name='cookie' direction='in'/>
+      <arg type='a(sa{sv})' name='identities' direction='in'/>
+    </method>
+    <method name='CancelAuthentication'>
+      <arg type='s' name='cookie' direction='in'/>
+    </method>
+  </interface>
+</node>
+"""
+
+
+def runtime_dir() -> Path:
+    base = Path(os.environ.get("XDG_RUNTIME_DIR", f"/run/user/{os.getuid()}")) / "panama-polkit"
+    base.mkdir(mode=0o700, parents=True, exist_ok=True)
+    # Enforced rather than assumed: an inherited directory with looser
+    # permissions would expose every request that passes through it.
+    os.chmod(base, 0o700)
+    return base
+
+
+def log(message: str) -> None:
+    print(f"panama-polkit-agent: {message}", file=sys.stderr, flush=True)
+
+
+class Agent:
+    def __init__(self) -> None:
+        self.bus = Gio.bus_get_sync(Gio.BusType.SYSTEM, None)
+        self.pending: dict[str, Path] = {}
+        self.loop = GLib.MainLoop()
+
+    # ── Registration ────────────────────────────────────────────────────────
+
+    def subject(self) -> GLib.Variant:
+        """This login session, which is what the agent authenticates for.
+
+        XDG_SESSION_ID is set for a login shell but NOT in the systemd user
+        environment, so running as a unit has to ask logind which session this
+        process belongs to rather than trusting the variable to be there.
+        """
+        session_id = os.environ.get("XDG_SESSION_ID", "")
+        if not session_id:
+            session_id = self.session_from_logind()
+        if not session_id:
+            raise RuntimeError("could not determine the login session to register for")
+        return GLib.Variant("(sa{sv})", ("unix-session",
+                                         {"session-id": GLib.Variant("s", session_id)}))
+
+    def session_from_logind(self) -> str:
+        """This user's graphical session, asked of logind.
+
+        Not GetSessionByPID: a systemd USER unit runs under [email protected],
+        which belongs to no login session at all -- which is exactly why
+        XDG_SESSION_ID is missing when started that way. The user object's
+        Display property names the graphical session, which is the one whose
+        authentications this agent should answer.
+        """
+        try:
+            user_path = self.bus.call_sync(
+                "org.freedesktop.login1", "/org/freedesktop/login1",
+                "org.freedesktop.login1.Manager", "GetUser",
+                GLib.Variant("(u)", (os.getuid(),)), GLib.VariantType("(o)"),
+                Gio.DBusCallFlags.NONE, 5000, None).unpack()[0]
+            display = self.bus.call_sync(
+                "org.freedesktop.login1", user_path,
+                "org.freedesktop.DBus.Properties", "Get",
+                GLib.Variant("(ss)", ("org.freedesktop.login1.User", "Display")),
+                GLib.VariantType("(v)"), Gio.DBusCallFlags.NONE, 5000, None).unpack()[0]
+            # (session-id, object-path)
+            return str(display[0]) if display and display[0] else ""
+        except Exception:  # noqa: BLE001 - no session is a real answer
+            return ""
+
+    def register(self) -> None:
+        node = Gio.DBusNodeInfo.new_for_xml(INTROSPECTION)
+        # The object MUST be exported on the same connection that registers the
+        # agent. polkitd calls BeginAuthentication back on the unique name that
+        # called RegisterAuthenticationAgent -- so exporting on the session bus
+        # while registering from the system bus leaves polkitd calling a path
+        # that does not exist, and every request fails as "Not authorized"
+        # without ever prompting.
+        self.bus.register_object(
+            AGENT_PATH, node.interfaces[0], self.on_call, None, None)
+        self.bus.call_sync(
+            AUTHORITY_NAME, AUTHORITY_PATH, AUTHORITY_INTERFACE,
+            "RegisterAuthenticationAgent",
+            GLib.Variant.new_tuple(self.subject(),
+                                   GLib.Variant("s", "en_US.UTF-8"),
+                                   GLib.Variant("s", AGENT_PATH)),
+            None, Gio.DBusCallFlags.NONE, 10000, None)
+        log("registered with polkitd")
+
+    def unregister(self) -> None:
+        try:
+            self.bus.call_sync(
+                AUTHORITY_NAME, AUTHORITY_PATH, AUTHORITY_INTERFACE,
+                "UnregisterAuthenticationAgent",
+                GLib.Variant.new_tuple(self.subject(), GLib.Variant("s", AGENT_PATH)),
+                None, Gio.DBusCallFlags.NONE, 5000, None)
+            log("unregistered")
+        except Exception:  # noqa: BLE001 - shutting down either way
+            pass
+
+    # ── The agent interface ─────────────────────────────────────────────────
+
+    def on_call(self, connection, sender, path, interface, method, parameters, invocation):
+        if method == "BeginAuthentication":
+            action_id, message, icon_name, details, cookie, identities = parameters.unpack()
+            self.begin(action_id, message, icon_name, cookie, identities, invocation)
+            return
+        if method == "CancelAuthentication":
+            (cookie,) = parameters.unpack()
+            self.cancel(cookie)
+            invocation.return_value(None)
+            return
+        invocation.return_error_literal(Gio.dbus_error_quark(),
+                                        Gio.DBusError.UNKNOWN_METHOD, "Unknown method")
+
+    def usernames(self, identities: list) -> list[str]:
+        """Who polkit will accept an answer from, as names a person recognizes."""
+        names = []
+        for kind, attributes in identities:
+            if kind != "unix-user":
+                continue
+            uid = attributes.get("uid")
+            if uid is None:
+                continue
+            try:
+                import pwd
+
+                names.append(pwd.getpwuid(int(uid)).pw_name)
+            except (KeyError, ValueError):
+                continue
+        return names
+
+    def begin(self, action_id, message, icon_name, cookie, identities, invocation) -> None:
+        names = self.usernames(identities)
+        if not names:
+            invocation.return_error_literal(
+                Gio.dbus_error_quark(), Gio.DBusError.FAILED,
+                "No user is allowed to authenticate this request.")
+            return
+
+        request_path = runtime_dir() / f"request-{os.getpid()}-{int(time.time() * 1000)}.json"
+        payload = {
+            "actionId": action_id,
+            "message": message,
+            "iconName": icon_name,
+            "cookie": cookie,
+            "users": names,
+            "preferred": os.environ.get("USER", names[0]),
+        }
+        # 0600 before anything is written to it, so the cookie is never briefly
+        # world-readable.
+        handle = os.open(request_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+        with os.fdopen(handle, "w", encoding="utf-8") as stream:
+            json.dump(payload, stream)
+
+        self.pending[cookie] = request_path
+        response_path = request_path.with_suffix(".response")
+
+        if not self.ask_shell(request_path):
+            self.finish(cookie, request_path, response_path)
+            invocation.return_error_literal(
+                Gio.dbus_error_quark(), Gio.DBusError.FAILED,
+                "The desktop could not show an authentication prompt.")
+            return
+
+        # Poll for the shell's answer rather than blocking the main loop, so a
+        # Cancel from polkit is still processed while a prompt is open.
+        started = time.monotonic()
+
+        def check() -> bool:
+            if cookie not in self.pending:
+                # Cancelled from the polkit side.
+                self.finish(cookie, request_path, response_path)
+                invocation.return_error_literal(
+                    Gio.dbus_error_quark(), Gio.DBusError.FAILED, "Cancelled.")
+                return False
+            if response_path.exists():
+                try:
+                    answer = json.loads(response_path.read_text(encoding="utf-8"))
+                except (OSError, json.JSONDecodeError):
+                    answer = {"result": "failed"}
+                self.finish(cookie, request_path, response_path)
+                if answer.get("result") == "ok":
+                    # The helper has already told polkitd. Returning normally is
+                    # what completes the request.
+                    invocation.return_value(None)
+                else:
+                    invocation.return_error_literal(
+                        Gio.dbus_error_quark(), Gio.DBusError.FAILED,
+                        "Authentication was not completed.")
+                return False
+            if time.monotonic() - started > PROMPT_TIMEOUT_SECONDS:
+                self.finish(cookie, request_path, response_path)
+                invocation.return_error_literal(
+                    Gio.dbus_error_quark(), Gio.DBusError.FAILED, "Timed out.")
+                return False
+            return True
+
+        GLib.timeout_add(150, check)
+
+    def cancel(self, cookie: str) -> None:
+        request_path = self.pending.pop(cookie, None)
+        if request_path is None:
+            return
+        # A marker rather than a deletion: the shell watches for this to close a
+        # prompt that polkit no longer wants an answer to.
+        try:
+            request_path.with_suffix(".cancelled").touch(mode=0o600)
+        except OSError:
+            pass
+
+    def finish(self, cookie: str, request_path: Path, response_path: Path) -> None:
+        self.pending.pop(cookie, None)
+        for path in (request_path, response_path, request_path.with_suffix(".cancelled")):
+            try:
+                path.unlink()
+            except OSError:
+                pass
+
+    def ask_shell(self, request_path: Path) -> bool:
+        """Tell the shell to prompt. Only the path travels, never the contents."""
+        try:
+            result = subprocess.run(
+                ["qs", "ipc", "call", "polkit", "begin", str(request_path)],
+                capture_output=True, text=True, timeout=15, check=False)
+        except (OSError, subprocess.TimeoutExpired):
+            return False
+        if result.returncode != 0:
+            log(f"the shell refused the prompt: {result.stderr.strip()[:120]}")
+            return False
+        return True
+
+    def run(self) -> int:
+        self.register()
+        for received in (signal.SIGINT, signal.SIGTERM):
+            GLib.unix_signal_add(GLib.PRIORITY_DEFAULT, received, self.stop)
+        try:
+            self.loop.run()
+        finally:
+            self.unregister()
+        return 0
+
+    def stop(self) -> bool:
+        self.loop.quit()
+        return False
+
+
+def main() -> int:
+    try:
+        return Agent().run()
+    except Exception as error:  # noqa: BLE001
+        log(f"could not start: {error}")
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())