1 files changed, 300 insertions, 0 deletions
diff --git a/working/keyring-crash/probe.py b/working/keyring-crash/probe.py
new file mode 100644
index 0000000..b18f7a7
--- /dev/null
+++ b/working/keyring-crash/probe.py
@@ -0,0 +1,300 @@
+#!/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())