451 lines
18 KiB
Python
Executable File
451 lines
18 KiB
Python
Executable File
#!/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:]))
|