Give identity its due: native enrollment, honest deletion, and sign-in that stays home
Claude-Session: https://claude.ai/code/session_01Ms2FbjQy31TVf3CEvQhGM8
This commit is contained in:
Gabriel Brown
@@ -12,22 +12,45 @@ turning individual services on and off, and removing an account. That is the
|
||||
whole Online Accounts panel apart from one OAuth handshake.
|
||||
|
||||
Adding an account is the part that splits. The daemon's AddAccount takes
|
||||
credentials as an ARGUMENT -- it stores them, it does not obtain them. For
|
||||
password-based providers (Nextcloud, IMAP, WebDAV) that is a username and a
|
||||
password, which a settings app can reasonably collect. For Google it is an OAuth
|
||||
token, and the code that runs that exchange lives in libgoa-backend, which
|
||||
Fedora ships without a GIR binding -- reachable from C only. Reimplementing it
|
||||
would mean our own Google client credentials. So Google sign-in is handed to
|
||||
GNOME's panel, and only the sign-in.
|
||||
credentials as an ARGUMENT -- it stores them, it does not obtain them, and every
|
||||
part of that storing happens inside goa-daemon, which does have the backend.
|
||||
For password-based providers (Nextcloud, IMAP) that argument is a username and a
|
||||
password, which a settings app can reasonably collect, so those are added here.
|
||||
For Google it is an OAuth token, and the code that runs that exchange lives in
|
||||
libgoa-backend, which Fedora ships without a GIR binding -- reachable from C
|
||||
only. Reimplementing it would mean our own Google client credentials. So OAuth
|
||||
sign-in is handed to GNOME's panel, and only the sign-in.
|
||||
|
||||
A password reaches this script on stdin and nowhere else. argv is world-readable
|
||||
through /proc, so a password passed as an argument is published to every process
|
||||
on the machine.
|
||||
|
||||
Usage:
|
||||
panama-accounts list
|
||||
panama-accounts list | snapshot
|
||||
panama-accounts set <object-path> <service> <true|false>
|
||||
panama-accounts remove <object-path>
|
||||
panama-accounts add-nextcloud SERVER USERNAME (password on stdin)
|
||||
panama-accounts add-imap EMAIL IMAP_HOST SMTP_HOST USERNAME (password on stdin)
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import urllib.parse
|
||||
|
||||
|
||||
class BoundaryError(RuntimeError):
|
||||
"""A user-visible validation or sign-in failure."""
|
||||
|
||||
|
||||
# A canned GOA, for contract runs: a JSON file in the shape `list` prints. With
|
||||
# it set, nothing here imports gi or touches the session bus, and every change a
|
||||
# verb would have made is appended to PANAMA_ACCOUNTS_LOG instead -- with the
|
||||
# password replaced by its length, so a contract can prove it was read from
|
||||
# stdin without a secret ever reaching a file.
|
||||
FIXTURE_ENV = "PANAMA_ACCOUNTS_FIXTURE"
|
||||
LOG_ENV = "PANAMA_ACCOUNTS_LOG"
|
||||
|
||||
# Every service GOA models. The account object carries one interface per service
|
||||
# it supports, so presence of the interface is what "this account can do mail"
|
||||
@@ -111,8 +134,319 @@ def find(client, path):
|
||||
return None
|
||||
|
||||
|
||||
# ── Adding a password account ────────────────────────────────────────────────
|
||||
#
|
||||
# The keys below are not invented. They are the keys goa-daemon writes into
|
||||
# ~/.config/goa-1.0/accounts.conf, which is to say the ones each provider's
|
||||
# build_object reads back -- taken from GOA 3.58's goaowncloudprovider.c and
|
||||
# goaimapsmtpprovider.c, and confirmed against the accounts already on this
|
||||
# machine. A key GOA does not recognize is silently ignored, so getting one
|
||||
# wrong produces an account that exists and does nothing.
|
||||
|
||||
|
||||
def read_password() -> str:
|
||||
"""The password, from stdin. Never an argument, at any point in the chain."""
|
||||
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.")
|
||||
return secret.decode("utf-8", "surrogateescape")
|
||||
|
||||
|
||||
# RFC 1123 hostname (or a dotted IPv4). These strings end up in accounts.conf
|
||||
# and in URIs other processes fetch, so a shell metacharacter or a space is a
|
||||
# malformed address whichever way it got here -- refuse it at the boundary.
|
||||
HOSTNAME = re.compile(
|
||||
r"^[A-Za-z0-9]([A-Za-z0-9-]{0,61}[A-Za-z0-9])?"
|
||||
r"(\.[A-Za-z0-9]([A-Za-z0-9-]{0,61}[A-Za-z0-9])?)*$")
|
||||
|
||||
|
||||
def valid_host(text: str) -> bool:
|
||||
return bool(text) and len(text) <= 253 and bool(HOSTNAME.match(text))
|
||||
|
||||
|
||||
def nextcloud_uris(server: str) -> tuple[str, str, str]:
|
||||
"""(WebDAV URI, DAV URI, host) for a Nextcloud address.
|
||||
|
||||
`remote.php/webdav` and `remote.php/dav` are Nextcloud's fixed layout, and
|
||||
are exactly what GOA's own provider derives from the server it was given.
|
||||
"""
|
||||
text = (server or "").strip()
|
||||
if not text:
|
||||
raise BoundaryError("Enter the address of your Nextcloud server.")
|
||||
if "://" not in text:
|
||||
text = "https://" + text
|
||||
|
||||
parsed = urllib.parse.urlsplit(text)
|
||||
if parsed.scheme not in ("http", "https"):
|
||||
raise BoundaryError("A server address starts with https://.")
|
||||
if not parsed.hostname or not valid_host(parsed.hostname):
|
||||
raise BoundaryError("That is not a server address.")
|
||||
# urlsplit is permissive about what follows the host; a path is fine, but
|
||||
# spaces or shell punctuation anywhere in the address are not an URL that
|
||||
# a Nextcloud server has.
|
||||
if re.search(r"[\s;|&`$<>\"']", text):
|
||||
raise BoundaryError("That is not a server address.")
|
||||
|
||||
base = f"{parsed.scheme}://{parsed.netloc}{parsed.path.rstrip('/')}"
|
||||
return base + "/remote.php/webdav/", base + "/remote.php/dav/", parsed.hostname
|
||||
|
||||
|
||||
def nextcloud_account(server: str, username: str) -> tuple[str, str, dict]:
|
||||
"""(identity, presentation identity, details). Validation and nothing else,
|
||||
so the same refusals run whether or not there is a GOA to talk to."""
|
||||
identity = (username or "").strip()
|
||||
if not identity:
|
||||
raise BoundaryError("Enter the user name for that server.")
|
||||
webdav, dav, host = nextcloud_uris(server)
|
||||
|
||||
return (
|
||||
identity,
|
||||
# What the account is shown as. A bare user name says nothing about
|
||||
# which server it is on, and people keep more than one.
|
||||
identity if "@" in identity else f"{identity}@{host}",
|
||||
{
|
||||
"Uri": webdav,
|
||||
"FilesEnabled": "true",
|
||||
"CalendarEnabled": "true",
|
||||
"CalDavUri": dav,
|
||||
"ContactsEnabled": "true",
|
||||
"CardDavUri": dav,
|
||||
"AcceptSslErrors": "false",
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def add_nextcloud(server: str, username: str, password: str) -> str:
|
||||
from gi.repository import GLib
|
||||
|
||||
identity, presentation, details = nextcloud_account(server, username)
|
||||
return add_account("owncloud", identity, presentation,
|
||||
{"password": GLib.Variant("s", password)},
|
||||
details, "Nextcloud")
|
||||
|
||||
|
||||
def imap_account(email: str, imap_host: str, smtp_host: str,
|
||||
username: str) -> tuple[str, str, dict]:
|
||||
address = (email or "").strip()
|
||||
if "@" not in address or address.startswith("@") or address.endswith("@"):
|
||||
raise BoundaryError("Enter the email address for this account.")
|
||||
imap = (imap_host or "").strip()
|
||||
smtp = (smtp_host or "").strip()
|
||||
if not imap or not smtp:
|
||||
raise BoundaryError("Enter both the incoming and outgoing server.")
|
||||
if not valid_host(imap) or not valid_host(smtp):
|
||||
raise BoundaryError("Mail servers are host names, like imap.example.org.")
|
||||
identity = (username or "").strip() or address
|
||||
|
||||
# The modern defaults, and the ones every mail provider this is likely to
|
||||
# meet uses: IMAP over TLS on 993, submission with STARTTLS on 587. GOA
|
||||
# offers a dropdown for the other combinations; a settings page that asked
|
||||
# four encryption questions to add a mailbox would be worse than one that
|
||||
# gets it right and lets GNOME's panel handle the unusual server.
|
||||
return (
|
||||
address, address,
|
||||
{
|
||||
"Enabled": "true",
|
||||
"EmailAddress": address,
|
||||
"Name": address,
|
||||
"ImapHost": imap,
|
||||
"ImapUserName": identity,
|
||||
"ImapUseSsl": "true",
|
||||
"ImapUseTls": "false",
|
||||
"ImapAcceptSslErrors": "false",
|
||||
"SmtpHost": smtp,
|
||||
"SmtpUseAuth": "true",
|
||||
"SmtpUserName": identity,
|
||||
"SmtpAuthLogin": "false",
|
||||
"SmtpAuthPlain": "true",
|
||||
"SmtpUseSsl": "false",
|
||||
"SmtpUseTls": "true",
|
||||
"SmtpAcceptSslErrors": "false",
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def add_imap(email: str, imap_host: str, smtp_host: str, username: str,
|
||||
password: str) -> str:
|
||||
from gi.repository import GLib
|
||||
|
||||
identity, presentation, details = imap_account(email, imap_host, smtp_host, username)
|
||||
# One password for both servers: submission almost always takes the same
|
||||
# credentials as the mailbox, and asking twice for the same secret is how a
|
||||
# form gets abandoned. GOA stores them under separate keys regardless.
|
||||
return add_account("imap_smtp", identity, presentation,
|
||||
{"imap-password": GLib.Variant("s", password),
|
||||
"smtp-password": GLib.Variant("s", password)},
|
||||
details, "mail")
|
||||
|
||||
|
||||
def add_account(provider: str, identity: str, presentation: str,
|
||||
credentials: dict, details: dict, label: str) -> str:
|
||||
"""Hand GOA a new account, then make it prove the credentials work.
|
||||
|
||||
The daemon does not check what it is given -- it writes the account out and
|
||||
stores the password. Without the second step a mistyped password produces an
|
||||
account that exists, looks fine, and never syncs, which is the failure this
|
||||
page is supposed to be able to explain.
|
||||
"""
|
||||
from gi.repository import GLib
|
||||
|
||||
client = load_client()
|
||||
manager = client.get_manager()
|
||||
if manager is None:
|
||||
raise BoundaryError("GNOME Online Accounts is not answering.")
|
||||
try:
|
||||
if not manager.call_is_supported_provider_sync(provider, None):
|
||||
raise BoundaryError(f"This system cannot add {label} accounts.")
|
||||
path = manager.call_add_account_sync(
|
||||
provider, identity, presentation,
|
||||
GLib.Variant("a{sv}", credentials),
|
||||
GLib.Variant("a{ss}", details), None)
|
||||
except GLib.Error as failure:
|
||||
raise BoundaryError(clean(failure.message)) from failure
|
||||
|
||||
verify(path, label)
|
||||
return path
|
||||
|
||||
|
||||
def verify(path: str, label: str) -> None:
|
||||
"""Sign in once, and undo the account if that fails."""
|
||||
from gi.repository import GLib
|
||||
|
||||
# A fresh client: the one that made the call has not necessarily seen the
|
||||
# object appear yet, and this is the cheapest way to wait for it properly.
|
||||
obj = find(load_client(), path)
|
||||
if obj is None:
|
||||
# It was added; this process simply cannot see it yet. Reporting a
|
||||
# failure here would be a lie, and the account list will show it.
|
||||
return
|
||||
|
||||
account = obj.get_account()
|
||||
try:
|
||||
account.set_default_timeout(60000)
|
||||
account.call_ensure_credentials_sync(None)
|
||||
except GLib.Error as failure:
|
||||
try:
|
||||
account.call_remove_sync(None)
|
||||
except GLib.Error:
|
||||
pass
|
||||
raise BoundaryError(
|
||||
f"Those {label} details were not accepted: {clean(failure.message)}"
|
||||
) from failure
|
||||
|
||||
|
||||
def clean(message: str) -> str:
|
||||
"""The useful sentence of a GOA error, without the D-Bus type prefix."""
|
||||
trimmed = re.sub(r"^GDBus\.Error:[^:]+:\s*", "", str(message or "")).strip()
|
||||
return trimmed.splitlines()[0][:200] if trimmed else "That did not work."
|
||||
|
||||
|
||||
# ── The canned GOA ───────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def fixture() -> dict | None:
|
||||
path = os.environ.get(FIXTURE_ENV)
|
||||
if not path:
|
||||
return None
|
||||
try:
|
||||
with open(path, encoding="utf-8") as handle:
|
||||
listing = json.load(handle)
|
||||
except (OSError, ValueError) as failure:
|
||||
return {"accounts": [], "error": f"The accounts fixture could not be read: {failure}"}
|
||||
listing.setdefault("accounts", [])
|
||||
listing.setdefault("error", "")
|
||||
return listing
|
||||
|
||||
|
||||
def record(verb: str, arguments: list, password: str | None = None) -> None:
|
||||
path = os.environ.get(LOG_ENV) or (os.environ.get(FIXTURE_ENV, "") + ".log")
|
||||
if not path:
|
||||
return
|
||||
entry = {"verb": verb, "arguments": arguments}
|
||||
if password is not None:
|
||||
# Its length, never the thing itself: the point of the log is to prove
|
||||
# the password came in on stdin, not to keep a copy of it.
|
||||
entry["passwordBytes"] = len(password)
|
||||
try:
|
||||
with open(path, "a", encoding="utf-8") as handle:
|
||||
handle.write(json.dumps(entry, separators=(",", ":")) + "\n")
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def added(operation) -> int:
|
||||
"""Run an add, and answer with the fresh account list either way.
|
||||
|
||||
Errors come back inside that list rather than on stderr, so the page has
|
||||
both facts -- what went wrong and what is there now -- from one read.
|
||||
"""
|
||||
error = ""
|
||||
try:
|
||||
operation()
|
||||
except BoundaryError as failure:
|
||||
error = str(failure)
|
||||
|
||||
try:
|
||||
accounts = [describe(obj) for obj in load_client().get_accounts()]
|
||||
except Exception: # noqa: BLE001 - the error above is the one worth saying
|
||||
accounts = []
|
||||
print(json.dumps({"accounts": accounts, "error": error}))
|
||||
return 0
|
||||
|
||||
|
||||
def replay(action: str, arguments: list, canned: dict) -> int:
|
||||
"""Every verb, against the canned GOA. No bus, no gi, no account changed."""
|
||||
error = ""
|
||||
try:
|
||||
if action in ("list", "snapshot"):
|
||||
pass
|
||||
elif action in ("set", "remove"):
|
||||
record(action, arguments[1:])
|
||||
elif action == "add-nextcloud":
|
||||
if len(arguments) != 3:
|
||||
raise BoundaryError("Usage: add-nextcloud SERVER USERNAME")
|
||||
password = read_password()
|
||||
nextcloud_account(arguments[1], arguments[2])
|
||||
record(action, arguments[1:], password)
|
||||
elif action == "add-imap":
|
||||
if len(arguments) != 5:
|
||||
raise BoundaryError("Usage: add-imap EMAIL IMAP_HOST SMTP_HOST USERNAME")
|
||||
password = read_password()
|
||||
imap_account(arguments[1], arguments[2], arguments[3], arguments[4])
|
||||
record(action, arguments[1:], password)
|
||||
else:
|
||||
raise BoundaryError(f"Unknown command {action!r}.")
|
||||
except BoundaryError as failure:
|
||||
error = str(failure)
|
||||
|
||||
print(json.dumps({**canned, "error": error or canned.get("error", "")}))
|
||||
return 0
|
||||
|
||||
|
||||
def main():
|
||||
action = sys.argv[1] if len(sys.argv) > 1 else "list"
|
||||
arguments = sys.argv[1:]
|
||||
action = arguments[0] if arguments else "list"
|
||||
|
||||
canned = fixture()
|
||||
if canned is not None:
|
||||
return replay(action, arguments, canned)
|
||||
|
||||
if action == "add-nextcloud":
|
||||
if len(arguments) != 3:
|
||||
print("usage: panama-accounts add-nextcloud SERVER USERNAME", file=sys.stderr)
|
||||
return 2
|
||||
return added(lambda: add_nextcloud(arguments[1], arguments[2], read_password()))
|
||||
|
||||
if action == "add-imap":
|
||||
if len(arguments) != 5:
|
||||
print("usage: panama-accounts add-imap EMAIL IMAP_HOST SMTP_HOST USERNAME",
|
||||
file=sys.stderr)
|
||||
return 2
|
||||
return added(lambda: add_imap(arguments[1], arguments[2], arguments[3],
|
||||
arguments[4], read_password()))
|
||||
|
||||
try:
|
||||
client = load_client()
|
||||
@@ -124,7 +458,9 @@ def main():
|
||||
}))
|
||||
return 0
|
||||
|
||||
if action == "list":
|
||||
# "snapshot" is what every other Panama helper calls this, and what the
|
||||
# service asks for; "list" is what this one has always been called.
|
||||
if action in ("list", "snapshot"):
|
||||
print(json.dumps({
|
||||
"accounts": [describe(obj) for obj in client.get_accounts()],
|
||||
"error": "",
|
||||
@@ -162,7 +498,8 @@ def main():
|
||||
obj.get_account().call_remove_sync(None)
|
||||
return 0
|
||||
|
||||
print("usage: panama-accounts [list|set|remove]", file=sys.stderr)
|
||||
print("usage: panama-accounts [list|snapshot|set|remove|add-nextcloud|add-imap]",
|
||||
file=sys.stderr)
|
||||
return 2
|
||||
|
||||
|
||||
|
||||
@@ -1,89 +1,490 @@
|
||||
#!/usr/bin/env bash
|
||||
#!/usr/bin/env python3
|
||||
|
||||
# Fingerprint state and the one privileged switch, for the Users page.
|
||||
#
|
||||
# Two independent facts make a working fingerprint login, and conflating them
|
||||
# is how the feature usually confuses people: fprintd must hold at least one
|
||||
# enrolled print (GNOME's Users panel owns that dialog, and Panama hands off
|
||||
# to it), and PAM must be told to ask the reader at all, which on Fedora is
|
||||
# authselect's `with-fingerprint` feature. This helper reports both and can
|
||||
# flip the second.
|
||||
#
|
||||
# Usage:
|
||||
# panama-fingerprint status -> {"reader":bool,"readerName":"","enrolled":[],"pamEnabled":bool,"error":""}
|
||||
# panama-fingerprint set-unlock on|off (prompts through panama-sudo/polkit)
|
||||
#
|
||||
# authselect is baseline Fedora (it manages PAM for the whole install), and
|
||||
# fprintd ships with Workstation; a machine with neither simply reports no
|
||||
# reader, which hides the card.
|
||||
"""Fingerprint login: the enrolled prints, and the one privileged switch.
|
||||
|
||||
set -uo pipefail
|
||||
Two independent facts make a working fingerprint login, and conflating them is
|
||||
how the feature usually confuses people. fprintd must hold at least one enrolled
|
||||
print, and PAM must be told to ask the reader at all, which on Fedora is
|
||||
authselect's `with-fingerprint` feature. This helper reports both, enrolls and
|
||||
removes prints, and can flip the second.
|
||||
|
||||
PANAMA_PATH="${PANAMA_PATH:-$HOME/.local/share/Panama}"
|
||||
Both facts are reported unconditionally, including on a machine with no reader.
|
||||
The state that used to be invisible -- the feature switched on with nothing
|
||||
enrolled and no reader attached -- is exactly the state someone needs to see and
|
||||
turn off, and reporting `pamEnabled: false` because there was no reader to ask
|
||||
made it unreachable.
|
||||
|
||||
emit() {
|
||||
jq -cn \
|
||||
--argjson reader "$1" \
|
||||
--arg readerName "$2" \
|
||||
--argjson enrolled "$3" \
|
||||
--argjson pamEnabled "$4" \
|
||||
--arg error "$5" \
|
||||
'{reader: $reader, readerName: $readerName, enrolled: $enrolled,
|
||||
pamEnabled: $pamEnabled, error: $error}'
|
||||
}
|
||||
Enrollment talks to fprintd over D-Bus (net.reactivated.Fprint) rather than
|
||||
handing the person to GNOME's Users panel: Claim, EnrollStart, then one
|
||||
`EnrollStatus` signal per touch until the device says it is done. Progress is
|
||||
printed as one JSON object per line, the same streaming shape panama-dictate's
|
||||
setup uses, so the page can count touches while they happen. That signal loop is
|
||||
why this is Python and no longer bash -- `status` and `set-unlock` still shell
|
||||
out to exactly the same tools, and answer in exactly the same shapes.
|
||||
|
||||
cmd_status() {
|
||||
command -v fprintd-list >/dev/null 2>&1 || { emit false "" '[]' false ""; return; }
|
||||
panama-fingerprint status
|
||||
panama-fingerprint set-unlock on|off (prompts through panama-sudo/polkit)
|
||||
panama-fingerprint enroll FINGER (streams {stage,done,total,result})
|
||||
panama-fingerprint remove FINGER
|
||||
panama-fingerprint remove-all
|
||||
|
||||
# fprintd-list both answers "is there a reader" (fprintd is bus-activated,
|
||||
# so this also copes with the daemon not running yet) and names the
|
||||
# enrolled fingers in one call.
|
||||
local listing
|
||||
# LC_ALL=C: the "no devices" match below reads fprintd's message, and a
|
||||
# translated daemon would turn every readerless non-English machine into
|
||||
# a permanent error card.
|
||||
if ! listing="$(LC_ALL=C timeout 10 fprintd-list "$USER" 2>&1)"; then
|
||||
# "No devices available" is the normal no-reader machine; anything
|
||||
# else is a real problem worth surfacing.
|
||||
if grep -qi 'no devices' <<<"$listing"; then
|
||||
emit false "" '[]' false ""
|
||||
else
|
||||
emit false "" '[]' false "fprintd did not answer: $(head -1 <<<"$listing")"
|
||||
fi
|
||||
authselect is baseline Fedora (it manages PAM for the whole install) and fprintd
|
||||
ships with Workstation; a machine with neither reports no reader and no feature.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
|
||||
FPRINT = "net.reactivated.Fprint"
|
||||
MANAGER_PATH = "/net/reactivated/Fprint/Manager"
|
||||
MANAGER_INTERFACE = "net.reactivated.Fprint.Manager"
|
||||
DEVICE_INTERFACE = "net.reactivated.Fprint.Device"
|
||||
|
||||
# The authselect feature that decides whether PAM asks the reader at unlock.
|
||||
FEATURE = "with-fingerprint"
|
||||
|
||||
# How long a reader may sit waiting for a finger before enrollment is given up
|
||||
# on. The page has a Cancel button; this is only for a session left open.
|
||||
IDLE_TIMEOUT_SECONDS = 90
|
||||
|
||||
# A canned fprintd, for contract runs. Points at a JSON file; see replay() for
|
||||
# the shape. Every call that would have gone to the bus is appended to
|
||||
# PANAMA_FINGERPRINT_LOG instead, so a test can pin the order of
|
||||
# Claim / EnrollStart / EnrollStop / Release without a reader in the room.
|
||||
FIXTURE_ENV = "PANAMA_FINGERPRINT_FIXTURE"
|
||||
LOG_ENV = "PANAMA_FINGERPRINT_LOG"
|
||||
|
||||
PANAMA_PATH = os.environ.get("PANAMA_PATH") or os.path.expanduser("~/.local/share/Panama")
|
||||
|
||||
# fprintd's own vocabulary, only far enough to reject nonsense before it becomes
|
||||
# a D-Bus error. What a finger is CALLED is presented by services/Fingerprint.qml,
|
||||
# which is the single place that decides how these read.
|
||||
FINGER = re.compile(r"^(left|right)-(thumb|(index|middle|ring|little)-finger)$")
|
||||
|
||||
|
||||
class BoundaryError(RuntimeError):
|
||||
"""A user-visible failure: no reader, a refused claim, a bad finger name."""
|
||||
|
||||
|
||||
# ── status ───────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def unlock_feature_enabled() -> bool:
|
||||
"""Whether PAM has been told to ask the reader.
|
||||
|
||||
Asked unconditionally. This is a property of the PAM configuration and has
|
||||
nothing to do with whether a reader is plugged in, which is the whole point:
|
||||
the feature left on with no reader is a state someone has to be able to see.
|
||||
"""
|
||||
if not shutil.which("authselect"):
|
||||
return False
|
||||
try:
|
||||
current = subprocess.run(["authselect", "current"], capture_output=True,
|
||||
text=True, timeout=10, check=False)
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
return False
|
||||
return FEATURE in (current.stdout or "")
|
||||
|
||||
|
||||
def status() -> dict:
|
||||
reader, name, enrolled, error = False, "", [], ""
|
||||
|
||||
if shutil.which("fprintd-list"):
|
||||
# fprintd-list answers "is there a reader" (fprintd is bus-activated, so
|
||||
# this copes with the daemon not running yet) and names the enrolled
|
||||
# fingers in one call.
|
||||
#
|
||||
# LC_ALL=C: the "no devices" match below reads fprintd's message, and a
|
||||
# translated daemon would turn every readerless non-English machine into
|
||||
# a permanent error card.
|
||||
environment = dict(os.environ, LC_ALL="C")
|
||||
try:
|
||||
listing = subprocess.run(["fprintd-list", os.environ.get("USER", "")],
|
||||
capture_output=True, text=True, timeout=10,
|
||||
check=False, env=environment)
|
||||
output = (listing.stdout or "") + (listing.stderr or "")
|
||||
except (OSError, subprocess.SubprocessError) as failure:
|
||||
listing, output = None, str(failure)
|
||||
|
||||
if listing is None or listing.returncode != 0:
|
||||
# "No devices available" is the normal no-reader machine; anything
|
||||
# else is a real problem worth surfacing.
|
||||
if "no devices" not in output.lower():
|
||||
first = next((line for line in output.splitlines() if line.strip()), "")
|
||||
error = f"fprintd did not answer: {first}"
|
||||
else:
|
||||
reader = True
|
||||
# "Fingerprints for user gib on FocalTech ... (press):" carries the
|
||||
# reader product name; " - #0: right-index-finger" the enrollment.
|
||||
match = re.search(r"^Fingerprints for user \S+ on (.*) \(\w*\):$",
|
||||
output, re.MULTILINE)
|
||||
name = match.group(1) if match else ""
|
||||
enrolled = re.findall(r"^ *- #\d+: (.+)$", output, re.MULTILINE)
|
||||
|
||||
feature = unlock_feature_enabled()
|
||||
return {
|
||||
"reader": reader,
|
||||
"readerName": name,
|
||||
"enrolled": enrolled,
|
||||
# Two names for one fact, on purpose: `pamEnabled` is what this helper
|
||||
# has always called it, `unlockFeatureEnabled` is what it is.
|
||||
"pamEnabled": feature,
|
||||
"unlockFeatureEnabled": feature,
|
||||
"error": error,
|
||||
}
|
||||
|
||||
|
||||
# ── the privileged switch ────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def set_unlock(state: str) -> int:
|
||||
if state == "on":
|
||||
verb = "enable-feature"
|
||||
reason = ("Turning on fingerprint login: telling PAM (via authselect) to "
|
||||
"ask the fingerprint reader when unlocking")
|
||||
elif state == "off":
|
||||
verb = "disable-feature"
|
||||
reason = ("Turning off fingerprint login: telling PAM (via authselect) to "
|
||||
"stop asking the fingerprint reader")
|
||||
else:
|
||||
print("panama-fingerprint set-unlock takes on|off", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
escalate = os.path.join(PANAMA_PATH, "bin", "panama-sudo")
|
||||
prefix = ([escalate, "--reason", reason, "--"]
|
||||
if os.access(escalate, os.X_OK) else ["sudo"])
|
||||
return subprocess.run(prefix + ["authselect", verb, FEATURE], check=False).returncode
|
||||
|
||||
|
||||
# ── talking to fprintd ───────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def emit(**fields) -> None:
|
||||
"""One JSON object, one line, flushed. The page reads these as they arrive."""
|
||||
print(json.dumps(fields, separators=(",", ":")), flush=True)
|
||||
|
||||
|
||||
def log_call(method: str, *arguments) -> None:
|
||||
"""Record a call the fixture stood in for, so a contract can pin the order."""
|
||||
path = os.environ.get(LOG_ENV) or (os.environ.get(FIXTURE_ENV, "") + ".log")
|
||||
if not path:
|
||||
return
|
||||
fi
|
||||
try:
|
||||
with open(path, "a", encoding="utf-8") as handle:
|
||||
handle.write(json.dumps({"method": method, "arguments": list(arguments)},
|
||||
separators=(",", ":")) + "\n")
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
# "Fingerprints for user gib on FocalTech ... (press):" carries the reader
|
||||
# product name; " - #0: right-index-finger" lines carry the enrollment.
|
||||
local name enrolled pam
|
||||
name="$(sed -n 's/^Fingerprints for user [^ ]* on \(.*\) (\w*):$/\1/p' <<<"$listing" | head -1)"
|
||||
enrolled="$(sed -n 's/^ *- #[0-9]*: //p' <<<"$listing" | jq -Rn '[inputs]')"
|
||||
pam=false
|
||||
authselect current 2>/dev/null | grep -q 'with-fingerprint' && pam=true
|
||||
|
||||
emit true "$name" "$enrolled" "$pam" ""
|
||||
}
|
||||
def fixture() -> dict | None:
|
||||
"""The canned fprintd, or None when there is a real bus to talk to.
|
||||
|
||||
cmd_set_unlock() {
|
||||
local verb reason
|
||||
case "$1" in
|
||||
on) verb=enable-feature
|
||||
reason="Turning on fingerprint login: telling PAM (via authselect) to ask the fingerprint reader when unlocking" ;;
|
||||
off) verb=disable-feature
|
||||
reason="Turning off fingerprint login: telling PAM (via authselect) to stop asking the fingerprint reader" ;;
|
||||
*) echo 'panama-fingerprint set-unlock takes on|off' >&2; exit 1 ;;
|
||||
esac
|
||||
{"enrollStages": 5,
|
||||
"results": ["enroll-stage-passed", "enroll-retry-scan-too-short", ...],
|
||||
"error": ""} <- non-empty stands in for a refused claim
|
||||
"""
|
||||
path = os.environ.get(FIXTURE_ENV)
|
||||
if not path:
|
||||
return None
|
||||
try:
|
||||
with open(path, encoding="utf-8") as handle:
|
||||
return json.load(handle)
|
||||
except (OSError, ValueError) as failure:
|
||||
raise BoundaryError(f"The fingerprint fixture could not be read: {failure}")
|
||||
|
||||
local sudo_cmd=(sudo)
|
||||
[[ -x "$PANAMA_PATH/bin/panama-sudo" ]] && sudo_cmd=(
|
||||
"$PANAMA_PATH/bin/panama-sudo" --reason "$reason" --
|
||||
)
|
||||
"${sudo_cmd[@]}" authselect "$verb" with-fingerprint
|
||||
}
|
||||
|
||||
case "${1:-}" in
|
||||
status) cmd_status ;;
|
||||
set-unlock) [[ -n "${2:-}" ]] || { echo 'panama-fingerprint set-unlock takes on|off' >&2; exit 1; }
|
||||
cmd_set_unlock "$2" ;;
|
||||
*) echo 'usage: panama-fingerprint status | set-unlock on|off' >&2; exit 1 ;;
|
||||
esac
|
||||
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 failure: # noqa: BLE001 - no bus is a legitimate state
|
||||
raise BoundaryError("The fingerprint service is not answering.") from failure
|
||||
|
||||
|
||||
def call(path: str, interface: str, method: str, parameters=None, reply=None):
|
||||
Gio, GLib, connection = bus()
|
||||
try:
|
||||
result = connection.call_sync(
|
||||
FPRINT, path, interface, method, parameters,
|
||||
GLib.VariantType(reply) if reply else None,
|
||||
Gio.DBusCallFlags.NONE, 30000, None)
|
||||
except Exception as failure: # noqa: BLE001
|
||||
raise BoundaryError(_clean(str(failure))) from failure
|
||||
return result.unpack() if result is not None else None
|
||||
|
||||
|
||||
def _clean(message: str) -> str:
|
||||
"""The useful sentence out of a D-Bus error, without the type prefix."""
|
||||
trimmed = re.sub(r"^GDBus\.Error:[^:]+:\s*", "", message).strip()
|
||||
if "no devices" in trimmed.lower():
|
||||
return "No fingerprint reader is connected."
|
||||
if "permission denied" in trimmed.lower() or "not authorized" in trimmed.lower():
|
||||
return "That was not authorized."
|
||||
if "already in use" in trimmed.lower() or "claimed" in trimmed.lower():
|
||||
return "The fingerprint reader is busy with something else."
|
||||
return trimmed.splitlines()[0][:200] if trimmed else "The fingerprint reader failed."
|
||||
|
||||
|
||||
def default_device() -> str:
|
||||
return call(MANAGER_PATH, MANAGER_INTERFACE, "GetDefaultDevice", None, "(o)")[0]
|
||||
|
||||
|
||||
def enroll_stages(device: str) -> int:
|
||||
Gio, GLib, connection = bus()
|
||||
try:
|
||||
result = connection.call_sync(
|
||||
FPRINT, device, "org.freedesktop.DBus.Properties", "Get",
|
||||
GLib.Variant("(ss)", (DEVICE_INTERFACE, "num-enroll-stages")),
|
||||
GLib.VariantType("(v)"), Gio.DBusCallFlags.NONE, 10000, None)
|
||||
return max(1, int(result.unpack()[0]))
|
||||
except Exception: # noqa: BLE001 - a device that will not say is not fatal
|
||||
# Readers overwhelmingly want five touches, and a counter that is wrong
|
||||
# is better than a page with no counter at all.
|
||||
return 5
|
||||
|
||||
|
||||
# ── enroll ───────────────────────────────────────────────────────────────────
|
||||
|
||||
# fprintd says "enroll-stage-passed" for a touch that counted and
|
||||
# "enroll-completed" when there are no more to take. Everything else beginning
|
||||
# "enroll-retry" or naming a placement problem is a touch to do again, and the
|
||||
# terminal failures arrive with done=true.
|
||||
STAGE_PASSED = "enroll-stage-passed"
|
||||
COMPLETED = "enroll-completed"
|
||||
|
||||
# The results that end an enrollment badly. The device says so itself over the
|
||||
# wire (the signal's `done` flag), so this list is only what the canned fprintd
|
||||
# has to recognize on its own.
|
||||
TERMINAL_FAILURES = (
|
||||
"enroll-failed", "enroll-data-full", "enroll-disconnected",
|
||||
"enroll-duplicate", "enroll-unknown-error",
|
||||
)
|
||||
|
||||
|
||||
def enroll(finger: str) -> int:
|
||||
if not FINGER.fullmatch(finger or ""):
|
||||
raise BoundaryError("That is not a finger fprintd knows.")
|
||||
|
||||
canned = fixture()
|
||||
if canned is not None:
|
||||
return enroll_replay(finger, canned)
|
||||
|
||||
device = default_device()
|
||||
total = enroll_stages(device)
|
||||
emit(ok=True, stage="claiming", done=0, total=total, result="", error="")
|
||||
|
||||
Gio, GLib, connection = bus()
|
||||
call(device, DEVICE_INTERFACE, "Claim",
|
||||
GLib.Variant("(s)", (os.environ.get("USER", ""),)))
|
||||
|
||||
loop = GLib.MainLoop()
|
||||
state = {"done": 0, "result": "", "error": "", "ok": False, "seen": time.monotonic()}
|
||||
|
||||
def on_status(_connection, _sender, _path, _interface, _signal, parameters):
|
||||
result, finished = parameters.unpack()
|
||||
state["seen"] = time.monotonic()
|
||||
state["result"] = result
|
||||
if result == STAGE_PASSED:
|
||||
state["done"] = min(state["done"] + 1, total)
|
||||
if finished:
|
||||
state["ok"] = result == COMPLETED
|
||||
if not state["ok"]:
|
||||
state["error"] = describe_result(result)
|
||||
loop.quit()
|
||||
return
|
||||
emit(ok=True, stage="scanning", done=state["done"], total=total,
|
||||
result=result, error="")
|
||||
|
||||
subscription = connection.signal_subscribe(
|
||||
None, DEVICE_INTERFACE, "EnrollStatus", device, None,
|
||||
Gio.DBusSignalFlags.NONE, on_status)
|
||||
|
||||
# A cancelled enrollment is a terminated process -- the page drops the
|
||||
# Process and Quickshell sends a signal. The device must still be released,
|
||||
# or the next attempt finds the reader busy with a session that is gone.
|
||||
def cancelled(_data=None):
|
||||
state["error"] = ""
|
||||
state["result"] = "cancelled"
|
||||
loop.quit()
|
||||
return GLib.SOURCE_REMOVE
|
||||
|
||||
GLib.unix_signal_add(GLib.PRIORITY_DEFAULT, 15, cancelled, None)
|
||||
GLib.unix_signal_add(GLib.PRIORITY_DEFAULT, 2, cancelled, None)
|
||||
|
||||
def watchdog(_data=None):
|
||||
if time.monotonic() - state["seen"] > IDLE_TIMEOUT_SECONDS:
|
||||
state["error"] = "The reader stopped answering."
|
||||
loop.quit()
|
||||
return GLib.SOURCE_REMOVE
|
||||
return GLib.SOURCE_CONTINUE
|
||||
|
||||
GLib.timeout_add_seconds(5, watchdog, None)
|
||||
|
||||
try:
|
||||
call(device, DEVICE_INTERFACE, "EnrollStart", GLib.Variant("(s)", (finger,)))
|
||||
emit(ok=True, stage="scanning", done=0, total=total, result="", error="")
|
||||
loop.run()
|
||||
finally:
|
||||
connection.signal_unsubscribe(subscription)
|
||||
# Both are best-effort: the interesting failure already happened, and a
|
||||
# reader left claimed is worse than a second error nobody can act on.
|
||||
for method in ("EnrollStop", "Release"):
|
||||
try:
|
||||
call(device, DEVICE_INTERFACE, method)
|
||||
except BoundaryError:
|
||||
pass
|
||||
|
||||
return finish(state, total)
|
||||
|
||||
|
||||
def finish(state: dict, total: int) -> int:
|
||||
if state["ok"]:
|
||||
emit(ok=True, stage="done", done=total, total=total,
|
||||
result=COMPLETED, error="")
|
||||
return 0
|
||||
stage = "cancelled" if state["result"] == "cancelled" else "failed"
|
||||
emit(ok=False, stage=stage, done=state["done"], total=total,
|
||||
result=state["result"], error=state["error"])
|
||||
return 1
|
||||
|
||||
|
||||
def describe_result(result: str) -> str:
|
||||
"""fprintd's terminal results, in words someone can act on."""
|
||||
return {
|
||||
"enroll-failed": "That finger could not be read. Try enrolling it again.",
|
||||
"enroll-data-full": "The reader has no room for another fingerprint.",
|
||||
"enroll-disconnected": "The fingerprint reader was disconnected.",
|
||||
"enroll-duplicate": "That finger is already enrolled.",
|
||||
"enroll-unknown-error": "The fingerprint reader failed.",
|
||||
}.get(result, "Enrolling that finger did not finish.")
|
||||
|
||||
|
||||
def enroll_replay(finger: str, canned: dict) -> int:
|
||||
"""The same stream, from a file. No bus, no reader, no waiting."""
|
||||
total = max(1, int(canned.get("enrollStages", 5)))
|
||||
emit(ok=True, stage="claiming", done=0, total=total, result="", error="")
|
||||
|
||||
log_call("Claim", os.environ.get("USER", ""))
|
||||
refusal = str(canned.get("error") or "")
|
||||
if refusal:
|
||||
emit(ok=False, stage="failed", done=0, total=total, result="", error=refusal)
|
||||
return 1
|
||||
|
||||
log_call("EnrollStart", finger)
|
||||
emit(ok=True, stage="scanning", done=0, total=total, result="", error="")
|
||||
|
||||
state = {"done": 0, "result": "", "error": "", "ok": False}
|
||||
for result in canned.get("results", []):
|
||||
state["result"] = result
|
||||
if result == STAGE_PASSED:
|
||||
state["done"] = min(state["done"] + 1, total)
|
||||
if result == COMPLETED or result in TERMINAL_FAILURES:
|
||||
state["ok"] = result == COMPLETED
|
||||
if not state["ok"]:
|
||||
state["error"] = describe_result(result)
|
||||
break
|
||||
emit(ok=True, stage="scanning", done=state["done"], total=total,
|
||||
result=result, error="")
|
||||
|
||||
log_call("EnrollStop")
|
||||
log_call("Release")
|
||||
return finish(state, total)
|
||||
|
||||
|
||||
# ── remove ───────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def remove(finger: str | None) -> dict:
|
||||
"""Delete one enrolled finger, or all of them, and report the fresh state."""
|
||||
if finger is not None and not FINGER.fullmatch(finger):
|
||||
raise BoundaryError("That is not a finger fprintd knows.")
|
||||
|
||||
canned = fixture()
|
||||
if canned is not None:
|
||||
log_call("Claim", os.environ.get("USER", ""))
|
||||
if finger is None:
|
||||
log_call("DeleteEnrolledFingers2")
|
||||
else:
|
||||
log_call("DeleteEnrolledFinger", finger)
|
||||
log_call("Release")
|
||||
return {**status(), "error": str(canned.get("error") or "")}
|
||||
|
||||
from gi.repository import GLib
|
||||
|
||||
device = default_device()
|
||||
call(device, DEVICE_INTERFACE, "Claim",
|
||||
GLib.Variant("(s)", (os.environ.get("USER", ""),)))
|
||||
try:
|
||||
if finger is None:
|
||||
# DeleteEnrolledFingers2 works on the claimed user; its predecessor
|
||||
# took a name and is deprecated for exactly the confusion that
|
||||
# invited -- deleting someone else's prints by typo.
|
||||
call(device, DEVICE_INTERFACE, "DeleteEnrolledFingers2")
|
||||
else:
|
||||
call(device, DEVICE_INTERFACE, "DeleteEnrolledFinger",
|
||||
GLib.Variant("(s)", (finger,)))
|
||||
finally:
|
||||
try:
|
||||
call(device, DEVICE_INTERFACE, "Release")
|
||||
except BoundaryError:
|
||||
pass
|
||||
|
||||
return status()
|
||||
|
||||
|
||||
# ── entry ────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def main(arguments: list[str]) -> int:
|
||||
verb = arguments[0] if arguments else ""
|
||||
|
||||
try:
|
||||
if verb == "status" and len(arguments) == 1:
|
||||
print(json.dumps(status(), separators=(",", ":")))
|
||||
return 0
|
||||
|
||||
if verb == "set-unlock" and len(arguments) == 2:
|
||||
return set_unlock(arguments[1])
|
||||
|
||||
if verb == "enroll" and len(arguments) == 2:
|
||||
return enroll(arguments[1])
|
||||
|
||||
if verb == "remove" and len(arguments) == 2:
|
||||
print(json.dumps(remove(arguments[1]), separators=(",", ":")))
|
||||
return 0
|
||||
|
||||
if verb == "remove-all" and len(arguments) == 1:
|
||||
print(json.dumps(remove(None), separators=(",", ":")))
|
||||
return 0
|
||||
|
||||
except BoundaryError as failure:
|
||||
# Enrollment streams, so its failure has to arrive in the same shape as
|
||||
# its progress; the others answer with the state plus the message, so a
|
||||
# page never has to ask twice to find out what happened.
|
||||
if verb == "enroll":
|
||||
emit(ok=False, stage="failed", done=0, total=0, result="",
|
||||
error=str(failure))
|
||||
else:
|
||||
print(json.dumps({**status(), "error": str(failure)},
|
||||
separators=(",", ":")))
|
||||
return 1
|
||||
|
||||
print("usage: panama-fingerprint status | set-unlock on|off | enroll FINGER | "
|
||||
"remove FINGER | remove-all", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main(sys.argv[1:]))
|
||||
|
||||
@@ -13,14 +13,17 @@ 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]
|
||||
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
|
||||
@@ -39,6 +42,16 @@ 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}$")
|
||||
|
||||
|
||||
@@ -137,6 +150,40 @@ def snapshot() -> dict:
|
||||
}
|
||||
|
||||
|
||||
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
|
||||
|
||||
@@ -195,6 +242,14 @@ def crop_square(path: str, x: int, y: int, size: int) -> str:
|
||||
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.")
|
||||
|
||||
@@ -221,7 +276,21 @@ def set_account_type(username: str, kind: str) -> None:
|
||||
|
||||
if kind not in ("standard", "administrator"):
|
||||
raise BoundaryError("That is not an account type.")
|
||||
call(user_path(username), USER_INTERFACE, "SetAccountType",
|
||||
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,)))
|
||||
|
||||
|
||||
@@ -232,6 +301,29 @@ def set_automatic_login(username: str, enabled: bool) -> None:
|
||||
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()
|
||||
@@ -269,10 +361,17 @@ def create_user(username: str, real_name: str, kind: str) -> None:
|
||||
"(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"):
|
||||
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.")
|
||||
@@ -285,7 +384,7 @@ def delete_user(username: str, files: str) -> None:
|
||||
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")))
|
||||
GLib.Variant("(xb)", (target["uid"], files in REMOVE_FILES)))
|
||||
|
||||
|
||||
def main(arguments: list[str]) -> int:
|
||||
@@ -294,6 +393,12 @@ def main(arguments: list[str]) -> int:
|
||||
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":
|
||||
@@ -308,19 +413,24 @@ def main(arguments: list[str]) -> int:
|
||||
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-accounts snapshot | set-real-name USER NAME | "
|
||||
"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-password USER | "
|
||||
"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-files|remove-files")
|
||||
"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.
|
||||
|
||||
Reference in New Issue
Block a user