#!/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 = """ """ def sweep_stale(base: Path) -> None: """Remove leftovers from an agent that did not exit cleanly. A killed agent leaves its request and response behind. They are harmless -- a spent cookie and a one-word result -- but they accumulate, and a directory of stale capabilities is a bad habit even when each one is inert. """ for path in base.glob("request-*"): try: if time.time() - path.stat().st_mtime > PROMPT_TIMEOUT_SECONDS: path.unlink() except OSError: continue 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) sweep_stale(base) 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 user@N.service, 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())