Debugging glib2 context sources

main.rs

use std::{time::Duration, thread::sleep};

use zbus::blocking::Connection;

mod mutter;

fn main() {
    let connection = Connection::session().unwrap();
    let proxy = mutter::IdleMonitorProxyBlocking::new(&connection).unwrap();
    for _ in 0..10000 {
        let res = proxy.add_idle_watch(110000 * 1000);
        // println!("{:?}", res);
    }
    loop {
        sleep(Duration::from_secs(10));
    }
}

mutter.rs

//! # DBus interface proxy for: `org.gnome.Mutter.IdleMonitor`
//!
//! This code was generated by `zbus-xmlgen` `3.0.0` from DBus introspection data.
//! Source: `Interface '/org/gnome/Mutter/IdleMonitor/Core' from service 'org.gnome.Mutter.IdleMonitor' on session bus`.
//!
//! You may prefer to adapt it, instead of using it verbatim.
//!
//! More information can be found in the
//! [Writing a client proxy](https://dbus.pages.freedesktop.org/zbus/client.html)
//! section of the zbus documentation.
//!
//! This DBus object implements
//! [standard DBus interfaces](https://dbus.freedesktop.org/doc/dbus-specification.html),
//! (`org.freedesktop.DBus.*`) for which the following zbus proxies can be used:
//!
//! * [`zbus::fdo::PropertiesProxy`]
//! * [`zbus::fdo::IntrospectableProxy`]
//! * [`zbus::fdo::PeerProxy`]
//!
//! …consequently `zbus-xmlgen` did not generate code for the above interfaces.

use zbus::dbus_proxy;

#[dbus_proxy(interface = "org.gnome.Mutter.IdleMonitor", default_path = "/org/gnome/Mutter/IdleMonitor/Core")]
trait IdleMonitor {
    /// AddIdleWatch method
    fn add_idle_watch(&self, interval: u64) -> zbus::Result<u32>;

    /// AddUserActiveWatch method
    fn add_user_active_watch(&self) -> zbus::Result<u32>;

    /// GetIdletime method
    fn get_idletime(&self) -> zbus::Result<u64>;

    /// RemoveWatch method
    fn remove_watch(&self, id: u32) -> zbus::Result<()>;

    /// ResetIdletime method
    fn reset_idletime(&self) -> zbus::Result<()>;

    /// WatchFired signal
    #[dbus_proxy(signal)]
    fn watch_fired(&self, id: u32) -> zbus::Result<()>;
}

payload.py

from pwn import *
import argparse

parser = argparse.ArgumentParser()
parser.add_argument('file', help='file to run')
args = parser.parse_args()

# context.log_level = "debug"
p = process(f'gdb -c {args.file}', shell=True)
p.recvuntil(">>>")
p.sendline("set style enabled off")
p.recvuntil(">>>")
p.sendline("up")
p.recvuntil(">>>")
# p.sendline("up")
# p.recvuntil(">>>")
p.sendline("print -elements unlimited -- context->sources")
data = p.recvuntil(">>>").decode("utf-8")

address = []

state = 0
number = ""
for i in range(len(data)):
    current_char = data[i]
    if current_char == "=":
        state = 1
    if current_char == "," or current_char == "[":
        state = 0
        if number:
            address.append(number)
            number = ""
    if state == 1:
        if current_char.isdigit() or current_char.isalpha():
            number += current_char

# print(address)

for i in address:
    p.sendline(f"print *(GSource *){i}")
    print(p.recvuntil(">>>").decode("utf-8"))
    p.sendline(f"print ((DBusWatch *)((MetaIdleMonitorWatch *)((GSourceCallback *)((GSource*){i})->callback_data)->data)->user_data)->dbus_name")
    print(p.recvuntil(">>>").decode("utf-8"))

p.interactive()