#!/usr/bin/env python3 """Secret Service stress probe for the disposable keyring guest only. Every mode drives real GIO D-Bus clients against org.freedesktop.secrets and reports what the daemon actually answered, not just that messages were sent. The guard at the top refuses to run anywhere but the disposable guest. Modes: plain OpenSession(plain) then Close, synchronous, one per client dh OpenSession(dh-ietf1024-sha256-aes128-cbc-pkcs7) then Close invalid OpenSession with an unsupported algorithm; expects NotSupported disconnect fire --burst OpenSession calls and drop the connection, no reply read pipeline fire --burst OpenSession calls, await and validate every reply, then Close every session alias a fresh client's first calls are Unlock, Lock and ReadAlias on the default alias path (the lazy client registration happens inside these calls; exercises the alias-string lifetime) isolation client A opens a session; client B must not be able to Close it; A then closes it and B's own session still works Each batch of --batch clients is followed by a daemon round trip (ReadAlias plus the collection's Locked property), so the counters reflect requests the daemon actually processed. Exit status is 1 on any error or on a change of bus owner. """ import argparse import collections import concurrent.futures import json import os import socket import sys import time import gi gi.require_version("Gio", "2.0") from gi.repository import Gio, GLib # noqa: E402 SERVICE = "org.freedesktop.secrets" SERVICE_PATH = "/org/freedesktop/secrets" SERVICE_IFACE = "org.freedesktop.Secret.Service" SESSION_IFACE = "org.freedesktop.Secret.Session" COLLECTION_IFACE = "org.freedesktop.Secret.Collection" ALIAS_PATH = "/org/freedesktop/secrets/aliases/default" LOGIN_PATH = "/org/freedesktop/secrets/collection/login" DH_ALGORITHM = "dh-ietf1024-sha256-aes128-cbc-pkcs7" TIMEOUT_MS = 10000 def connect(): conn = Gio.DBusConnection.new_for_address_sync( os.environ["DBUS_SESSION_BUS_ADDRESS"], Gio.DBusConnectionFlags.AUTHENTICATION_CLIENT | Gio.DBusConnectionFlags.MESSAGE_BUS_CONNECTION, None, None, ) conn.set_exit_on_close(False) return conn def call(conn, path, iface, method, args, reply_type): return conn.call_sync( SERVICE, path, iface, method, args, GLib.VariantType.new(reply_type) if reply_type else None, Gio.DBusCallFlags.NONE, TIMEOUT_MS, None, ) def owner(conn): return conn.call_sync( "org.freedesktop.DBus", "/org/freedesktop/DBus", "org.freedesktop.DBus", "GetNameOwner", GLib.Variant("(s)", (SERVICE,)), GLib.VariantType.new("(s)"), Gio.DBusCallFlags.NONE, 5000, None, ).unpack()[0] def barrier(conn): """Await the daemon itself, not only the message bus, after each batch.""" path = call(conn, SERVICE_PATH, SERVICE_IFACE, "ReadAlias", GLib.Variant("(s)", ("default",)), "(o)").unpack()[0] if path != LOGIN_PATH: raise RuntimeError("Synthetic login collection disappeared") locked = conn.call_sync( SERVICE, path, "org.freedesktop.DBus.Properties", "Get", GLib.Variant("(ss)", (COLLECTION_IFACE, "Locked")), GLib.VariantType.new("(v)"), Gio.DBusCallFlags.NONE, TIMEOUT_MS, None, ).unpack()[0] if locked: raise RuntimeError("Synthetic login collection became locked") def open_session_body(mode): algorithm = "plain" if mode in ("plain", "disconnect", "pipeline", "isolation") else DH_ALGORITHM payload = GLib.Variant("s", "") if algorithm == "plain" else GLib.Variant("ay", [2]) if mode == "invalid": algorithm = "intentionally-unsupported" return GLib.Variant("(sv)", (algorithm, payload)) def check_open_reply(reply): """An OpenSession reply must carry a session object path under the service.""" _output, path = reply.unpack() if not isinstance(path, str) or not path.startswith(SERVICE_PATH + "/session/"): raise RuntimeError(f"OpenSession returned an invalid session path: {path!r}") return path def close_session(conn, path): call(conn, path, SESSION_IFACE, "Close", None, None) def run_sync_client(conn, mode): try: reply = call(conn, SERVICE_PATH, SERVICE_IFACE, "OpenSession", open_session_body(mode), "(vo)") except GLib.Error as exc: if mode == "invalid" and Gio.DBusError.get_remote_error(exc) == "org.freedesktop.DBus.Error.NotSupported": return "expected-not-supported" raise if mode == "invalid": raise RuntimeError("Unsupported algorithm unexpectedly accepted") close_session(conn, check_open_reply(reply)) return "ok" def run_disconnect_client(conn, burst): for _ in range(burst): message = Gio.DBusMessage.new_method_call(SERVICE, SERVICE_PATH, SERVICE_IFACE, "OpenSession") message.set_body(open_session_body("disconnect")) conn.send_message(message, Gio.DBusSendMessageFlags.NONE) conn.flush_sync(None) return "ok" def run_pipeline_client(conn, burst): """Queue --burst calls without waiting, then collect and validate every reply.""" loop = GLib.MainLoop() pending = {"left": burst, "paths": [], "errors": []} def done(source, result): try: pending["paths"].append(check_open_reply(source.call_finish(result))) except Exception as exc: # noqa: BLE001 - recorded, then raised below pending["errors"].append(str(exc)) pending["left"] -= 1 if pending["left"] == 0: loop.quit() for _ in range(burst): conn.call(SERVICE, SERVICE_PATH, SERVICE_IFACE, "OpenSession", open_session_body("pipeline"), GLib.VariantType.new("(vo)"), Gio.DBusCallFlags.NONE, TIMEOUT_MS, None, done) GLib.timeout_add(TIMEOUT_MS + 2000, loop.quit) loop.run() if pending["left"]: raise RuntimeError(f"{pending['left']} pipelined OpenSession calls never answered") if pending["errors"]: raise RuntimeError("pipelined OpenSession failed: " + pending["errors"][0]) if len(set(pending["paths"])) != burst: raise RuntimeError("pipelined sessions did not get distinct paths") for path in pending["paths"]: close_session(conn, path) return "ok" def run_alias_client(conn): """First calls on a fresh connection all go through the alias path. Unlock on an already-unlocked collection is a no-op that still resolves the alias inside the daemon (the alias-in-arguments path the message filter does not rewrite). Lock is deliberately not used: it would lock the fixture's login collection, and unlocking again needs a password prompt the headless guest cannot show. """ unlocked, prompt = call(conn, SERVICE_PATH, SERVICE_IFACE, "Unlock", GLib.Variant("(ao)", ([ALIAS_PATH],)), "(aoo)").unpack() if unlocked != [ALIAS_PATH] or prompt != "/": raise RuntimeError(f"Unlock via alias answered {unlocked!r}, prompt {prompt!r}") path = call(conn, SERVICE_PATH, SERVICE_IFACE, "ReadAlias", GLib.Variant("(s)", ("default",)), "(o)").unpack()[0] if path != LOGIN_PATH: raise RuntimeError(f"ReadAlias answered {path!r}") # A property read addressed to the alias path: the filter rewrites the # object path, then the property getter looks the collection up. locked = conn.call_sync( SERVICE, ALIAS_PATH, "org.freedesktop.DBus.Properties", "Get", GLib.Variant("(ss)", (COLLECTION_IFACE, "Locked")), GLib.VariantType.new("(v)"), Gio.DBusCallFlags.NONE, TIMEOUT_MS, None, ).unpack()[0] if locked: raise RuntimeError("login collection reported locked through the alias path") return "ok" def run_isolation_client(conn_a): conn_b = connect() try: path_a = check_open_reply(call(conn_a, SERVICE_PATH, SERVICE_IFACE, "OpenSession", open_session_body("isolation"), "(vo)")) path_b = check_open_reply(call(conn_b, SERVICE_PATH, SERVICE_IFACE, "OpenSession", open_session_body("isolation"), "(vo)")) if path_a == path_b: raise RuntimeError("two clients received the same session path") try: close_session(conn_b, path_a) except GLib.Error: pass else: raise RuntimeError("client B was allowed to Close client A's session") # A's session must still be alive after B's attempt: a second Close by A # must succeed once (the session exists), and B's own session still closes. close_session(conn_a, path_a) close_session(conn_b, path_b) return "ok" finally: conn_b.close_sync(None) def run_client(mode, burst): conn = connect() try: if mode in ("plain", "dh", "invalid"): return run_sync_client(conn, mode) if mode == "disconnect": return run_disconnect_client(conn, burst) if mode == "pipeline": return run_pipeline_client(conn, burst) if mode == "alias": return run_alias_client(conn) if mode == "isolation": return run_isolation_client(conn) raise ValueError(mode) finally: conn.close_sync(None) def daemon_rss_kib(conn): """RSS of the current bus owner, read via its PID; None when unavailable.""" try: pid = conn.call_sync( "org.freedesktop.DBus", "/org/freedesktop/DBus", "org.freedesktop.DBus", "GetConnectionUnixProcessID", GLib.Variant("(s)", (SERVICE,)), GLib.VariantType.new("(u)"), Gio.DBusCallFlags.NONE, 5000, None, ).unpack()[0] with open(f"/proc/{pid}/status") as status: for line in status: if line.startswith("VmRSS:"): return int(line.split()[1]) except (GLib.Error, OSError): return None return None def main(): parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("mode", choices=("plain", "dh", "invalid", "disconnect", "pipeline", "alias", "isolation")) parser.add_argument("--clients", type=int, default=200) parser.add_argument("--workers", type=int, default=16) parser.add_argument("--burst", type=int, default=1) parser.add_argument("--batch", type=int, default=100) args = parser.parse_args() if socket.gethostname() != "archsetup-test" or os.getuid() != 1000: raise SystemExit("Refusing to run outside the disposable archsetup-test guest as UID 1000") monitor = connect() initial_owner = owner(monitor) rss_before = daemon_rss_kib(monitor) results = collections.Counter() examples = [] start = time.monotonic() with concurrent.futures.ThreadPoolExecutor(max_workers=args.workers) as pool: for offset in range(0, args.clients, args.batch): futures = [pool.submit(run_client, args.mode, args.burst) for _ in range(min(args.batch, args.clients - offset))] for future in concurrent.futures.as_completed(futures): try: results[future.result()] += 1 except Exception as exc: # noqa: BLE001 - counted and reported results[type(exc).__name__] += 1 if len(examples) < 5: examples.append(str(exc)) barrier(monitor) try: final_owner = owner(monitor) except GLib.Error as exc: final_owner = str(exc) rss_after = daemon_rss_kib(monitor) monitor.close_sync(None) print(json.dumps({ "mode": args.mode, "clients": args.clients, "workers": args.workers, "burst": args.burst, "seconds": round(time.monotonic() - start, 2), "initial_owner": initial_owner, "final_owner": final_owner, "rss_kib_before": rss_before, "rss_kib_after": rss_after, "results": dict(results), "errors": examples, }), flush=True) return 1 if examples or initial_owner != final_owner else 0 if __name__ == "__main__": sys.exit(main())