#!/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-users snapshot
    panama-users stock-avatars
    panama-users set-real-name USER NAME
    panama-users set-icon USER PATH [X Y SIZE]     (PATH "" clears the picture)
    panama-users set-account-type USER standard|administrator
    panama-users set-automatic-login USER true|false
    panama-users set-locked USER true|false
    panama-users set-password USER                 (new password on stdin)
    panama-users reset-password USER               (no password material at all)
    panama-users create-user USERNAME REALNAME standard|administrator
    panama-users delete-user USERNAME keep|remove
"""

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

# Password modes, likewise. 1 is "the account has no usable password and must
# choose one at the next sign-in" -- which is why resetting a password here
# involves no password at all, not even one this process saw for a moment.
PASSWORD_MODE_SET_AT_LOGIN = 1

# Where a distribution keeps the pictures its login screen offers. Fedora ships
# fifteen; a machine with none is normal and yields an empty list.
STOCK_AVATAR_DIR = "/usr/share/pixmaps/faces"
STOCK_AVATAR_SUFFIXES = (".png", ".jpg", ".jpeg", ".webp", ".svg")

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 stock_avatars() -> dict:
    """The pictures the distribution ships, as {name, path} pairs.

    Absolute paths, because the caller draws them straight from disk and then
    hands the same path back to set-icon. Sorted by name so the gallery does
    not reshuffle itself between openings.
    """
    entries = []
    try:
        names = os.listdir(STOCK_AVATAR_DIR)
    except OSError:
        # No such directory is an ordinary machine, not a failure worth a row.
        names = []

    for name in names:
        if not name.lower().endswith(STOCK_AVATAR_SUFFIXES):
            continue
        path = os.path.join(STOCK_AVATAR_DIR, name)
        if not os.path.isfile(path):
            continue
        entries.append({"name": _avatar_label(name), "path": path})

    entries.sort(key=lambda entry: entry["name"].lower())
    return {"avatars": entries, "error": ""}


def _avatar_label(filename: str) -> str:
    """"coffee2.jpg" -> "Coffee 2". A file name is not a caption, but it is the
    only thing these pictures carry, so it is tidied rather than invented."""
    stem = os.path.splitext(filename)[0]
    words = re.sub(r"(\d+)$", r" \1", stem.replace("-", " ").replace("_", " ")).strip()
    return words[:1].upper() + words[1:]


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

    # An empty path is how accountsservice is told to forget the picture: the
    # same call, with nothing in it. There is no separate "clear" method, and
    # inventing a verb for it here would only hide that.
    if path == "":
        call(user_path(username), USER_INTERFACE, "SetIconFile",
             GLib.Variant("(s)", ("",)))
        return

    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.")
    path = user_path(username)

    # The same reason deleting the last administrator is refused: a machine
    # whose only administrator has just been demoted cannot be administered,
    # and the demotion itself is the last thing that needed authorization.
    if kind == "standard":
        state = snapshot()
        target = next((user for user in state["users"]
                       if user["userName"] == username), None)
        if target is not None and target["administrator"] \
                and state["administratorCount"] <= 1:
            raise BoundaryError(
                "That is the only administrator; the machine would have none.")

    call(path, 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_locked(username: str, locked: bool) -> None:
    """Lock or unlock an account. A locked account cannot sign in at all, which
    is what someone is looking at when a user row says nothing works for them."""
    from gi.repository import GLib

    call(user_path(username), USER_INTERFACE, "SetLocked",
         GLib.Variant("(b)", (locked,)))


def reset_password(username: str) -> None:
    """Require a new password at the next sign-in.

    Deliberately not "set a password for them": no password is chosen, typed,
    hashed, or transmitted. accountsservice is told the account's password mode
    is "set at login", and the login screen collects the new one from the person
    who is going to use it.
    """
    from gi.repository import GLib

    call(user_path(username), USER_INTERFACE, "SetPasswordMode",
         GLib.Variant("(i)", (PASSWORD_MODE_SET_AT_LOGIN,)))


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)")


# What to do with the home directory, spelled either way. "keep"/"remove" is
# what the page says out loud; the longer pair is what this helper has always
# taken, and callers older than the page still pass it.
KEEP_FILES = ("keep", "keep-files")
REMOVE_FILES = ("remove", "remove-files")


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 in REMOVE_FILES)))


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

        # Its own shape rather than a snapshot: this asks the filesystem what
        # pictures exist, which has nothing to do with who has an account.
        if arguments == ["stock-avatars"]:
            print(json.dumps(stock_avatars(), 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) == 3 and arguments[0] == "set-locked":
            set_locked(arguments[1], arguments[2] == "true")
        elif len(arguments) == 2 and arguments[0] == "set-password":
            set_password(arguments[1])
        elif len(arguments) == 2 and arguments[0] == "reset-password":
            reset_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-users snapshot | stock-avatars | 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-locked USER true|false | "
                "set-password USER | reset-password USER | "
                "create-user USERNAME REALNAME standard|administrator | "
                "delete-user USERNAME keep|remove")
    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:]))
