Panama/config/dot/quickshell/scripts/panama-polkit-agent

#!/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 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 [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())