path: root/working/keyring-crash/probe.py

#!/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())