491 lines
20 KiB
Python
Executable File
491 lines
20 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
"""Fingerprint login: the enrolled prints, and the one privileged switch.
|
|
|
|
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.
|
|
|
|
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.
|
|
|
|
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.
|
|
|
|
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
|
|
|
|
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
|
|
try:
|
|
with open(path, "a", encoding="utf-8") as handle:
|
|
handle.write(json.dumps({"method": method, "arguments": list(arguments)},
|
|
separators=(",", ":")) + "\n")
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def fixture() -> dict | None:
|
|
"""The canned fprintd, or None when there is a real bus to talk to.
|
|
|
|
{"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}")
|
|
|
|
|
|
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:]))
|