"""AI-generated, not human-reviewed. Disposable Secret Service race reproducer.

Uses only fictitious credentials. Never connects to the user's session bus.
No claim of an officially reviewed fix or exhaustive regression coverage.
"""

import argparse
import json
import resource
import signal
import subprocess
import tempfile
import time
from pathlib import Path

import gi

gi.require_version("Gio", "2.0")
from gi.repository import Gio, GLib


SERVICE = "org.freedesktop.secrets"
COLLECTION = "/org/freedesktop/secrets/collection/login"
FLAGS = (
    Gio.DBusConnectionFlags.AUTHENTICATION_CLIENT
    | Gio.DBusConnectionFlags.MESSAGE_BUS_CONNECTION
)


def connect(address):
    return Gio.DBusConnection.new_for_address_sync(address, FLAGS, None, None)


def run(mode, count, binary):
    resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
    with tempfile.TemporaryDirectory(prefix="cofre-repro-") as temporary:
        root = Path(temporary)
        for directory in ("home", "data", "runtime", "control"):
            (root / directory).mkdir(mode=0o700)
        address = f"unix:path={root}/bus"
        config = root / "bus.conf"
        config.write_text(
            "<busconfig><type>session</type>"
            f"<listen>{address}</listen>"
            '<auth>EXTERNAL</auth><policy context="default">'
            '<allow send_destination="*"/><allow receive_sender="*"/>'
            '<allow own="*"/></policy></busconfig>'
        )
        env = {
            "PATH": "/usr/bin:/bin",
            "HOME": str(root / "home"),
            "XDG_DATA_HOME": str(root / "data"),
            "XDG_RUNTIME_DIR": str(root / "runtime"),
            "DBUS_SESSION_BUS_ADDRESS": address,
            "LANG": "C.UTF-8",
        }
        bus = None
        daemon = None
        observer = None
        with (root / "daemon.log").open("w+") as log:
            try:
                bus = subprocess.Popen(
                    ["/usr/bin/dbus-daemon", "--nofork", "--nopidfile", f"--config-file={config}"],
                    env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                )
                for _ in range(100):
                    if (root / "bus").exists():
                        break
                    time.sleep(0.02)
                daemon = subprocess.Popen(
                    [
                        binary, "--foreground", "--unlock",
                        "--components=secrets", f"--control-directory={root}/control",
                    ],
                    env=env, stdin=subprocess.PIPE, stdout=log, stderr=log,
                )
                daemon.stdin.write(b"senha-ficticia-descartavel-046\n")
                daemon.stdin.close()
                observer = connect(address)
                for _ in range(100):
                    result = observer.call_sync(
                        "org.freedesktop.DBus", "/org/freedesktop/DBus",
                        "org.freedesktop.DBus", "NameHasOwner",
                        GLib.Variant("(s)", (SERVICE,)), None,
                        Gio.DBusCallFlags.NONE, 1000, None,
                    )
                    if result.unpack()[0]:
                        break
                    time.sleep(0.02)
                alias = observer.call_sync(
                    SERVICE, "/org/freedesktop/secrets",
                    "org.freedesktop.Secret.Service", "ReadAlias",
                    GLib.Variant("(s)", ("login",)), None,
                    Gio.DBusCallFlags.NONE, 1000, None,
                ).unpack()[0]
                if alias != COLLECTION:
                    raise RuntimeError(f"Unexpected disposable alias: {alias}")
                errors = 0
                attempts = 0
                paused = False
                for index in range(count):
                    if daemon.poll() is not None:
                        break
                    attempts += 1
                    if mode == "queued" and index % 50 == 0:
                        daemon.send_signal(signal.SIGSTOP)
                        paused = True
                    if mode == "busctl":
                        completed = subprocess.run(
                            [
                                "/usr/bin/busctl", "--user", "--auto-start=no",
                                "get-property", SERVICE, COLLECTION,
                                "org.freedesktop.Secret.Collection", "Locked",
                            ],
                            env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                            timeout=3,
                        )
                        errors += completed.returncode != 0
                    else:
                        connection = connect(address) if mode != "persistent" else observer
                        try:
                            if mode in ("disconnect", "queued"):
                                message = Gio.DBusMessage.new_method_call(
                                    SERVICE, COLLECTION, "org.freedesktop.DBus.Properties", "Get"
                                )
                                message.set_body(GLib.Variant(
                                    "(ss)", ("org.freedesktop.Secret.Collection", "Locked")
                                ))
                                connection.send_message(message, Gio.DBusSendMessageFlags.NONE)
                                connection.flush_sync(None)
                            else:
                                connection.call_sync(
                                    SERVICE, COLLECTION, "org.freedesktop.DBus.Properties", "Get",
                                    GLib.Variant(
                                        "(ss)", ("org.freedesktop.Secret.Collection", "Locked")
                                    ),
                                    None, Gio.DBusCallFlags.NONE, 1000, None,
                                )
                        except GLib.Error:
                            errors += 1
                        finally:
                            if connection is not observer:
                                connection.close_sync(None)
                    if mode == "queued" and index % 50 == 49:
                        daemon.send_signal(signal.SIGCONT)
                        paused = False
                        time.sleep(0.02)
                    if index % 50 == 0:
                        time.sleep(0.001)
                if paused and daemon.poll() is None:
                    daemon.send_signal(signal.SIGCONT)
                time.sleep(0.2)
                log.flush()
                log.seek(0)
                text = log.read()
                healthy = False
                locked_after = False
                if daemon.poll() is None:
                    healthy = observer.call_sync(
                        SERVICE, COLLECTION, "org.freedesktop.DBus.Properties", "Get",
                        GLib.Variant("(ss)", ("org.freedesktop.Secret.Collection", "Locked")),
                        None, Gio.DBusCallFlags.NONE, 1000, None,
                    ).unpack()[0] is False
                    observer.call_sync(
                        SERVICE, "/org/freedesktop/secrets", "org.freedesktop.Secret.Service",
                        "Lock", GLib.Variant("(ao)", ([COLLECTION],)),
                        None, Gio.DBusCallFlags.NONE, 1000, None,
                    )
                    locked_after = observer.call_sync(
                        SERVICE, COLLECTION, "org.freedesktop.DBus.Properties", "Get",
                        GLib.Variant("(ss)", ("org.freedesktop.Secret.Collection", "Locked")),
                        None, Gio.DBusCallFlags.NONE, 1000, None,
                    ).unpack()[0] is True
                result = {
                    "mode": mode, "attempts": attempts, "errors": errors,
                    "daemon_exit": daemon.poll(),
                    "missing_client": "assertion 'client' failed" in text,
                    "missing_session": "assertion 'session' failed" in text,
                    "glib_assertion": "error != NULL" in text,
                    "functional_after": healthy,
                    "locked_after": locked_after,
                    "log_tail": text[-2200:] if daemon.poll() is not None else "",
                }
                print(json.dumps(result), flush=True)
                return result
            finally:
                if observer is not None and not observer.is_closed():
                    observer.close_sync(None)
                for process in (daemon, bus):
                    if process is not None and process.poll() is None:
                        process.send_signal(signal.SIGCONT)
                        process.send_signal(signal.SIGTERM)
                        try:
                            process.wait(timeout=3)
                        except subprocess.TimeoutExpired:
                            process.kill()
                            process.wait(timeout=3)


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("mode", choices=("persistent", "sync", "busctl", "disconnect", "queued"))
    parser.add_argument("--count", type=int, default=500)
    parser.add_argument("--daemon", default="/usr/bin/gnome-keyring-daemon")
    parser.add_argument("--repeat", type=int, default=1)
    parser.add_argument("--expect-crash", action="store_true",
                        help="Controle negativo: exige pelo menos um aborto com a assinatura conhecida")
    args = parser.parse_args()
    if args.count < 1 or args.repeat < 1 or not Path(args.daemon).is_absolute():
        parser.error("count/repeat devem ser positivos e daemon deve ser absoluto")
    results = [run(args.mode, args.count, args.daemon) for _ in range(args.repeat)]
    if args.expect_crash:
        passed = any(result["daemon_exit"] == -signal.SIGABRT and result["missing_client"]
                     and result["missing_session"] and result["glib_assertion"] for result in results)
    else:
        passed = all(result["daemon_exit"] is None and result["functional_after"]
                     and result["locked_after"] and result["errors"] == 0
                     and not result["glib_assertion"] and not result["missing_session"]
                     and not result["missing_client"] for result in results)
    raise SystemExit(0 if passed else 1)