Panama/config/dot/quickshell/scripts/panama-users

#!/usr/bin/env python3

"""User accounts, through accountsservice -- the same daemon GNOME's Users panel
drives.

Everything here that changes something is authorized by polkit, which prompts
through the agent this desktop already runs. This script never asks for a
password itself and never holds an administrator credential.

The one credential it does handle is a NEW password being set. That is read
from stdin, hashed by openssl reading its own stdin, and handed to
accountsservice over D-Bus from inside this process. It is never an argument:
argv is world-readable through /proc, so a password -- or even its hash --
passed that way is published to every process on the machine.

    panama-accounts snapshot
    panama-accounts set-real-name USER NAME
    panama-accounts set-icon USER PATH [X Y SIZE]
    panama-accounts set-account-type USER standard|administrator
    panama-accounts set-automatic-login USER true|false
    panama-accounts set-password USER          (new password on stdin)
    panama-accounts create-user USERNAME REALNAME standard|administrator
    panama-accounts delete-user USERNAME [keep-files|remove-files]
"""

from __future__ import annotations

import json
import os
import re
import subprocess
import tempfile
import sys

ACCOUNTS = "org.freedesktop.Accounts"
ACCOUNTS_PATH = "/org/freedesktop/Accounts"
USER_INTERFACE = "org.freedesktop.Accounts.User"

# Account types as accountsservice numbers them.
STANDARD, ADMINISTRATOR = 0, 1

USERNAME = re.compile(r"^[a-z_][a-z0-9_-]{0,31}$")


class BoundaryError(RuntimeError):
    """A user-visible validation or permission failure."""


def bus():
    try:
        import gi

        gi.require_version("Gio", "2.0")
        from gi.repository import Gio, GLib

        return Gio, GLib, Gio.bus_get_sync(Gio.BusType.SYSTEM, None)
    except Exception as error:  # noqa: BLE001 - no bus is a legitimate state
        raise BoundaryError("The account service is not answering.") from error


def call(path: str, interface: str, method: str, parameters=None, reply=None):
    Gio, GLib, connection = bus()
    try:
        result = connection.call_sync(
            ACCOUNTS, path, interface, method, parameters,
            GLib.VariantType(reply) if reply else None,
            Gio.DBusCallFlags.NONE, 120000, None)
    except Exception as error:  # noqa: BLE001
        message = str(error)
        # Polkit's own refusal is the common case and deserves plain words
        # rather than a D-Bus error string.
        if "not authorized" in message.lower() or "dismissed" in message.lower():
            raise BoundaryError("That change was not authorized.") from error
        raise BoundaryError(_clean(message)) from error
    return result.unpack() if result is not None else None


def _clean(message: str) -> str:
    """The last useful sentence of a D-Bus error, without the type prefix."""
    trimmed = re.sub(r"^GDBus\.Error:[^:]+:\s*", "", message).strip()
    return trimmed.splitlines()[0][:200] if trimmed else "That change failed."


def properties(path: str) -> dict:
    Gio, GLib, connection = bus()
    result = connection.call_sync(
        ACCOUNTS, path, "org.freedesktop.DBus.Properties", "GetAll",
        GLib.Variant("(s)", (USER_INTERFACE,)), GLib.VariantType("(a{sv})"),
        Gio.DBusCallFlags.NONE, 20000, None)
    return result.unpack()[0]


def user_path(username: str) -> str:
    if not USERNAME.fullmatch(username or ""):
        raise BoundaryError("That is not a user name.")
    from gi.repository import GLib

    return call(ACCOUNTS_PATH, ACCOUNTS, "FindUserByName",
                GLib.Variant("(s)", (username,)), "(o)")[0]


def describe(path: str) -> dict:
    values = properties(path)
    icon = str(values.get("IconFile") or "")
    return {
        "path": path,
        "userName": str(values.get("UserName") or ""),
        "realName": str(values.get("RealName") or ""),
        # Reported only when it is actually there: accountsservice keeps the
        # path in its database whether or not a file exists, so a deleted
        # avatar otherwise shows as a broken image.
        "iconFile": icon if icon and os.path.isfile(icon) else "",
        "administrator": int(values.get("AccountType") or 0) == ADMINISTRATOR,
        "locked": bool(values.get("Locked")),
        "automaticLogin": bool(values.get("AutomaticLogin")),
        "loginTime": int(values.get("LoginTime") or 0),
        "shell": str(values.get("Shell") or ""),
        "homeDirectory": str(values.get("HomeDirectory") or ""),
        "systemAccount": bool(values.get("SystemAccount")),
        "uid": int(values.get("Uid") or 0),
    }


def snapshot() -> dict:
    paths = call(ACCOUNTS_PATH, ACCOUNTS, "ListCachedUsers", None, "(ao)")[0]
    users = [describe(path) for path in paths]
    users = [user for user in users if not user["systemAccount"]]
    users.sort(key=lambda user: user["uid"])
    me = os.environ.get("USER") or ""
    return {
        "users": users,
        "currentUser": me,
        # Removing the only administrator would leave a machine nobody can
        # administer, so the page needs to know rather than find out.
        "administratorCount": sum(1 for user in users if user["administrator"]),
        "error": "",
    }


def set_real_name(username: str, name: str) -> None:
    from gi.repository import GLib

    if len(name) > 128 or "\n" in name or ":" in name:
        raise BoundaryError("That name cannot be used.")
    call(user_path(username), USER_INTERFACE, "SetRealName",
         GLib.Variant("(s)", (name,)))


# What an avatar is written out at. accountsservice stores whatever it is
# handed, so a 4000px photograph would sit on disk forever to be drawn at 48.
AVATAR_SIZE = 512


def crop_square(path: str, x: int, y: int, size: int) -> str:
    """Cut a square out of a picture and write it at avatar size.

    Returns a path to a new file; the caller is responsible for removing it.
    GdkPixbuf rather than a new dependency -- gi is already required here for
    accountsservice itself.
    """
    import gi
    gi.require_version("GdkPixbuf", "2.0")
    from gi.repository import GdkPixbuf

    try:
        picture = GdkPixbuf.Pixbuf.new_from_file(path)
    except Exception as error:
        raise BoundaryError("That file is not a picture this can read.") from error

    if size <= 0:
        raise BoundaryError("That is not a region of the picture.")

    # Clamped rather than rejected: a drag that ends a pixel outside the image
    # is a normal thing to do with a pointer, not an error worth refusing.
    x = max(0, min(x, picture.get_width() - 1))
    y = max(0, min(y, picture.get_height() - 1))
    size = min(size, picture.get_width() - x, picture.get_height() - y)
    if size <= 0:
        raise BoundaryError("That region is outside the picture.")

    square = picture.new_subpixbuf(x, y, size, size)
    scaled = square.scale_simple(AVATAR_SIZE, AVATAR_SIZE, GdkPixbuf.InterpType.BILINEAR)
    if scaled is None:
        raise BoundaryError("That picture could not be resized.")

    handle, out = tempfile.mkstemp(prefix="panama-avatar-", suffix=".png")
    os.close(handle)
    # accountsservice reads this as root and copies it; it must not be private
    # to this user, and it is deleted as soon as that copy has happened.
    os.chmod(out, 0o644)
    scaled.savev(out, "png", [], [])
    return out


def set_icon(username: str, path: str, region: tuple[int, int, int] | None = None) -> None:
    from gi.repository import GLib

    if not os.path.isfile(path):
        raise BoundaryError("That picture no longer exists.")

    source = path
    temporary = None
    if region is not None:
        temporary = source = crop_square(path, *region)

    try:
        call(user_path(username), USER_INTERFACE, "SetIconFile",
             GLib.Variant("(s)", (source,)))
    finally:
        # SetIconFile copies the file before it returns, so this is safe here
        # and leaving it behind would litter /tmp with every picture ever set.
        if temporary is not None:
            try:
                os.unlink(temporary)
            except OSError:
                pass


def set_account_type(username: str, kind: str) -> None:
    from gi.repository import GLib

    if kind not in ("standard", "administrator"):
        raise BoundaryError("That is not an account type.")
    call(user_path(username), USER_INTERFACE, "SetAccountType",
         GLib.Variant("(i)", (ADMINISTRATOR if kind == "administrator" else STANDARD,)))


def set_automatic_login(username: str, enabled: bool) -> None:
    from gi.repository import GLib

    call(user_path(username), USER_INTERFACE, "SetAutomaticLogin",
         GLib.Variant("(b)", (enabled,)))


def set_password(username: str) -> None:
    """Set a new password, read from stdin and never named on a command line."""
    secret = sys.stdin.buffer.read()
    # A trailing newline from a pipe is not part of the password.
    if secret.endswith(b"\n"):
        secret = secret[:-1]
    if not secret:
        raise BoundaryError("No password was provided.")
    if len(secret) < 6:
        raise BoundaryError("That password is too short.")

    # openssl reads the password on ITS stdin too, so the cleartext never
    # appears in a process listing at any point in the chain.
    hashed = subprocess.run(["openssl", "passwd", "-6", "-stdin"],
                            input=secret, capture_output=True, timeout=30, check=False)
    if hashed.returncode != 0 or not hashed.stdout.strip():
        raise BoundaryError("The password could not be prepared.")

    from gi.repository import GLib

    call(user_path(username), USER_INTERFACE, "SetPassword",
         GLib.Variant("(ss)", (hashed.stdout.decode().strip(), "")))


def create_user(username: str, real_name: str, kind: str) -> None:
    from gi.repository import GLib

    if not USERNAME.fullmatch(username or ""):
        raise BoundaryError("A user name may use lowercase letters, digits, - and _.")
    if kind not in ("standard", "administrator"):
        raise BoundaryError("That is not an account type.")
    call(ACCOUNTS_PATH, ACCOUNTS, "CreateUser",
         GLib.Variant("(ssi)", (username, real_name,
                                ADMINISTRATOR if kind == "administrator" else STANDARD)),
         "(o)")


def delete_user(username: str, files: str) -> None:
    from gi.repository import GLib

    if files not in ("keep-files", "remove-files"):
        raise BoundaryError("Say whether to keep or remove the home directory.")
    if username == (os.environ.get("USER") or ""):
        raise BoundaryError("You cannot delete the account you are signed in to.")

    state = snapshot()
    target = next((user for user in state["users"] if user["userName"] == username), None)
    if target is None:
        raise BoundaryError("That account no longer exists.")
    if target["administrator"] and state["administratorCount"] <= 1:
        raise BoundaryError("That is the only administrator; the machine would have none.")

    call(ACCOUNTS_PATH, ACCOUNTS, "DeleteUser",
         GLib.Variant("(xb)", (target["uid"], files == "remove-files")))


def main(arguments: list[str]) -> int:
    try:
        if arguments == ["snapshot"]:
            print(json.dumps(snapshot(), separators=(",", ":")))
            return 0

        if len(arguments) == 3 and arguments[0] == "set-real-name":
            set_real_name(arguments[1], arguments[2])
        elif len(arguments) == 3 and arguments[0] == "set-icon":
            set_icon(arguments[1], arguments[2])
        elif len(arguments) == 6 and arguments[0] == "set-icon":
            try:
                region = tuple(int(value) for value in arguments[3:6])
            except ValueError:
                raise BoundaryError("That is not a region of the picture.")
            set_icon(arguments[1], arguments[2], region)
        elif len(arguments) == 3 and arguments[0] == "set-account-type":
            set_account_type(arguments[1], arguments[2])
        elif len(arguments) == 3 and arguments[0] == "set-automatic-login":
            set_automatic_login(arguments[1], arguments[2] == "true")
        elif len(arguments) == 2 and arguments[0] == "set-password":
            set_password(arguments[1])
        elif len(arguments) == 4 and arguments[0] == "create-user":
            create_user(arguments[1], arguments[2], arguments[3])
        elif len(arguments) == 3 and arguments[0] == "delete-user":
            delete_user(arguments[1], arguments[2])
        else:
            raise BoundaryError(
                "Usage: panama-accounts snapshot | set-real-name USER NAME | "
                "set-icon USER PATH [X Y SIZE] | set-account-type USER standard|administrator | "
                "set-automatic-login USER true|false | set-password USER | "
                "create-user USERNAME REALNAME standard|administrator | "
                "delete-user USERNAME keep-files|remove-files")
    except BoundaryError as error:
        # Answers with the fresh state plus the message, so a page never has to
        # ask twice to find out what happened.
        try:
            state = snapshot()
        except BoundaryError:
            state = {"users": [], "currentUser": "", "administratorCount": 0}
        state["error"] = str(error)
        print(json.dumps(state, separators=(",", ":")))
        return 0

    print(json.dumps(snapshot(), separators=(",", ":")))
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))