#!/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:]))