We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Refactor EventWrapper
It seems we can maybe remove EventWrapper class now since we depend on GDB >= 9.2?

https://github.com/pwndbg/pwndbg/issues/1854#issuecomment-1664527390

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `pwndbg/gdblib/events.py`
Content:
```
1 """
2 Enables callbacks into functions to be automatically invoked
3 when various events occur to the debuggee (e.g. STOP on SIGINT)
4 by using a decorator.
5 """
6 
7 from __future__ import annotations
8 
9 import sys
10 from functools import partial
11 from functools import wraps
12 from typing import Any
13 from typing import Callable
14 
15 import gdb
16 
17 from pwndbg.gdblib.config import config
18 
19 debug = config.add_param("debug-events", False, "display internal event debugging info")
20 
21 
22 # There is no GDB way to get a notification when the binary itself
23 # is loaded from disk, by the operating system, before absolutely
24 # anything happens
25 #
26 # However, we get an Objfile event when the binary is loaded, before
27 # its entry point is invoked.
28 #
29 # We also get an Objfile event when we load up GDB, so we need
30 # to detect when the binary is running or not.
31 #
32 # Additionally, when attaching to a process running under QEMU, the
33 # very first event which is fired is a 'stop' event.  We need to
34 # capture this so that we can fire off all of the 'start' events first.
35 class StartEvent:
36     def __init__(self) -> None:
37         self.registered: list[Callable] = []
38         self.running = False
39 
40     def connect(self, function) -> None:
41         if function not in self.registered:
42             self.registered.append(function)
43 
44     def disconnect(self, function) -> None:
45         if function in self.registered:
46             self.registered.remove(function)
47 
48     def on_new_objfile(self) -> None:
49         if self.running or not gdb.selected_thread():
50             return
51 
52         self.running = True
53 
54         for function in self.registered:
55             if debug:
56                 sys.stdout.write(
57                     "{!r} {}.{}\n".format("start", function.__module__, function.__name__)
58                 )
59             function()
60 
61     def on_exited(self) -> None:
62         self.running = False
63 
64     def on_stop(self) -> None:
65         self.on_new_objfile()
66 
67 
68 gdb.events.start = StartEvent()
69 
70 
71 class EventWrapper:
72     """
73     Wrapper for GDB events which may not exist on older GDB versions but we still can
74     fire them manually (to invoke them you have to call `invoke_callbacks`).
75     """
76 
77     def __init__(self, name: str) -> None:
78         self.name = name
79 
80         self._event = getattr(gdb.events, self.name, None)
81         self._is_real_event = self._event is not None
82 
83     def connect(self, func) -> None:
84         if self._event is not None:
85             self._event.connect(func)
86 
87     def disconnect(self, func) -> None:
88         if self._event is not None:
89             self._event.disconnect(func)
90 
91     @property
92     def is_real_event(self) -> bool:
93         return self._is_real_event
94 
95     def invoke_callbacks(self) -> None:
96         """
97         As an optimization please don't call this if your GDB has this event (check `.is_real_event`).
98         """
99         for f in registered[self]:
100             f()
101 
102 
103 # Old GDBs doesn't have gdb.events.before_prompt, so we will emulate it using gdb.prompt_hook
104 before_prompt_event = EventWrapper("before_prompt")
105 gdb.events.before_prompt = before_prompt_event
106 
107 
108 # In order to support reloading, we must be able to re-fire
109 # all 'objfile' and 'stop' events.
110 registered: dict[Any, list[Callable]] = {
111     gdb.events.exited: [],
112     gdb.events.cont: [],
113     gdb.events.new_objfile: [],
114     gdb.events.stop: [],
115     gdb.events.start: [],
116     gdb.events.new_thread: [],
117     gdb.events.before_prompt: [],  # The real event might not exist, but we wrap it
118     gdb.events.memory_changed: [],
119     gdb.events.register_changed: [],
120 }
121 
122 
123 # When performing remote debugging, gdbserver is very noisy about which
124 # objects are loaded.  This greatly slows down the debugging session.
125 # In order to combat this, we keep track of which objfiles have been loaded
126 # this session, and only emit objfile events for each *new* file.
127 objfile_cache: dict[str, set[str]] = {}
128 
129 
130 def connect(func, event_handler, name=""):
131     if debug:
132         print("Connecting", func.__name__, event_handler)
133 
134     @wraps(func)
135     def caller(*a):
136         if debug:
137             sys.stdout.write(f"{name!r} {func.__module__}.{func.__name__} {a!r}\n")
138 
139         if a and isinstance(a[0], gdb.NewObjFileEvent):
140             objfile = a[0].new_objfile
141             handler = f"{func.__module__}.{func.__name__}"
142             path = objfile.filename
143             dispatched = objfile_cache.get(path, set())
144 
145             if handler in dispatched:
146                 return
147 
148             dispatched.add(handler)
149             objfile_cache[path] = dispatched
150 
151         try:
152             func()
153         except Exception as e:
154             import pwndbg.exception
155 
156             pwndbg.exception.handle()
157             raise e
158 
159     registered[event_handler].append(caller)
160     event_handler.connect(caller)
161     return func
162 
163 
164 def exit(func):
165     return connect(func, gdb.events.exited, "exit")
166 
167 
168 def cont(func):
169     return connect(func, gdb.events.cont, "cont")
170 
171 
172 def new_objfile(func):
173     return connect(func, gdb.events.new_objfile, "obj")
174 
175 
176 def stop(func):
177     return connect(func, gdb.events.stop, "stop")
178 
179 
180 def start(func):
181     return connect(func, gdb.events.start, "start")
182 
183 
184 def thread(func):
185     return connect(func, gdb.events.new_thread, "thread")
186 
187 
188 before_prompt = partial(connect, event_handler=gdb.events.before_prompt, name="before_prompt")
189 
190 
191 def reg_changed(func):
192     try:
193         return connect(func, gdb.events.register_changed, "reg_changed")
194     except AttributeError:
195         return func
196 
197 
198 def mem_changed(func):
199     try:
200         return connect(func, gdb.events.memory_changed, "mem_changed")
201     except AttributeError:
202         return func
203 
204 
205 def log_objfiles(ofile=None) -> None:
206     if not (debug and ofile):
207         return
208 
209     name = ofile.new_objfile.filename
210 
211     print("objfile: %r" % name)
212     gdb.execute("info sharedlibrary")
213 
214 
215 gdb.events.new_objfile.connect(log_objfiles)
216 
217 
218 def after_reload(start=True) -> None:
219     if gdb.selected_inferior().pid:
220         for f in registered[gdb.events.stop]:
221             f()
222         for f in registered[gdb.events.start]:
223             if start:
224                 f()
225         for f in registered[gdb.events.new_objfile]:
226             f()
227         for f in registered[gdb.events.before_prompt]:
228             f()
229 
230 
231 def on_reload() -> None:
232     for event, functions in registered.items():
233         for function in functions:
234             event.disconnect(function)
235         registered[event] = []
236 
237 
238 @new_objfile
239 def _start_newobjfile() -> None:
240     gdb.events.start.on_new_objfile()
241 
242 
243 @exit
244 def _start_exit() -> None:
245     gdb.events.start.on_exited()
246 
247 
248 @stop
249 def _start_stop() -> None:
250     gdb.events.start.on_stop()
251 
252 
253 @exit
254 def _reset_objfiles() -> None:
255     global objfile_cache
256     objfile_cache = {}
257
```

Path: `pwndbg/gdblib/prompt.py`
Content:
```
1 from __future__ import annotations
2 
3 from os import environ
4 
5 import gdb
6 
7 import pwndbg.decorators
8 import pwndbg.gdblib.events
9 import pwndbg.gdblib.functions
10 import pwndbg.lib.cache
11 import pwndbg.profiling
12 from pwndbg.color import disable_colors
13 from pwndbg.color import message
14 from pwndbg.lib.tips import color_tip
15 from pwndbg.lib.tips import get_tip_of_the_day
16 
17 funcs_list_str = ", ".join(message.notice("$" + f.name) for f in pwndbg.gdblib.functions.functions)
18 
19 num_pwndbg_cmds = sum(
20     1 for _ in filter(lambda c: not (c.shell or c.is_alias), pwndbg.commands.commands)
21 )
22 num_shell_cmds = sum(1 for _ in filter(lambda c: c.shell, pwndbg.commands.commands))
23 hint_lines = (
24     "loaded %i pwndbg commands and %i shell commands. Type %s for a list."
25     % (num_pwndbg_cmds, num_shell_cmds, message.notice("pwndbg [--shell | --all] [filter]")),
26     f"created {funcs_list_str} GDB functions (can be used with print/break)",
27 )
28 
29 for line in hint_lines:
30     print(message.prompt("pwndbg: ") + message.system(line))
31 
32 # noinspection PyPackageRequirements
33 show_tip = pwndbg.gdblib.config.add_param(
34     "show-tips", True, "whether to display the tip of the day on startup"
35 )
36 
37 cur = None
38 
39 
40 def initial_hook(*a) -> None:
41     if show_tip and not pwndbg.decorators.first_prompt:
42         colored_tip = color_tip(get_tip_of_the_day())
43         print(
44             message.prompt("------- tip of the day")
45             + message.system(" (disable with %s)" % message.notice("set show-tips off"))
46             + message.prompt(" -------")
47         )
48         print(colored_tip)
49     pwndbg.decorators.first_prompt = True
50 
51     prompt_hook(*a)
52 
53     if environ.get("PWNDBG_PROFILE") == "1":
54         pwndbg.profiling.profiler.stop("pwndbg-first-prompt.pstats")
55 
56     gdb.prompt_hook = prompt_hook
57 
58 
59 context_shown = False
60 
61 
62 def prompt_hook(*a) -> None:
63     global cur, context_shown
64 
65     new = (gdb.selected_inferior(), gdb.selected_thread())
66 
67     if cur != new:
68         pwndbg.gdblib.events.after_reload(start=cur is None)
69         cur = new
70 
71     if pwndbg.gdblib.proc.alive and pwndbg.gdblib.proc.thread_is_stopped and not context_shown:
72         pwndbg.commands.context.context()
73         context_shown = True
74 
75 
76 @pwndbg.gdblib.events.cont
77 def reset_context_shown(*a) -> None:
78     global context_shown
79     context_shown = False
80 
81 
82 @pwndbg.gdblib.config.trigger(message.config_prompt_color, disable_colors)
83 def set_prompt() -> None:
84     prompt = "pwndbg> "
85 
86     if not disable_colors:
87         prompt = "\x02" + prompt + "\x01"  # STX + prompt + SOH
88         prompt = message.prompt(prompt)
89         prompt = "\x01" + prompt + "\x02"  # SOH + prompt + STX
90 
91     gdb.execute(f"set prompt {prompt}")
92 
93 
94 if pwndbg.gdblib.events.before_prompt_event.is_real_event:
95     gdb.prompt_hook = initial_hook
96 
97 else:
98     # Old GDBs doesn't have gdb.events.before_prompt, so we will emulate it using gdb.prompt_hook
99     def extended_prompt_hook(*a):
100         pwndbg.gdblib.events.before_prompt_event.invoke_callbacks()
101         return prompt_hook(*a)
102 
103     gdb.prompt_hook = extended_prompt_hook
104
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/pwndbg/gdblib/events.py b/pwndbg/gdblib/events.py
--- a/pwndbg/gdblib/events.py
+++ b/pwndbg/gdblib/events.py
@@ -67,44 +67,6 @@
 
 gdb.events.start = StartEvent()
 
-
-class EventWrapper:
-    """
-    Wrapper for GDB events which may not exist on older GDB versions but we still can
-    fire them manually (to invoke them you have to call `invoke_callbacks`).
-    """
-
-    def __init__(self, name: str) -> None:
-        self.name = name
-
-        self._event = getattr(gdb.events, self.name, None)
-        self._is_real_event = self._event is not None
-
-    def connect(self, func) -> None:
-        if self._event is not None:
-            self._event.connect(func)
-
-    def disconnect(self, func) -> None:
-        if self._event is not None:
-            self._event.disconnect(func)
-
-    @property
-    def is_real_event(self) -> bool:
-        return self._is_real_event
-
-    def invoke_callbacks(self) -> None:
-        """
-        As an optimization please don't call this if your GDB has this event (check `.is_real_event`).
-        """
-        for f in registered[self]:
-            f()
-
-
-# Old GDBs doesn't have gdb.events.before_prompt, so we will emulate it using gdb.prompt_hook
-before_prompt_event = EventWrapper("before_prompt")
-gdb.events.before_prompt = before_prompt_event
-
-
 # In order to support reloading, we must be able to re-fire
 # all 'objfile' and 'stop' events.
 registered: dict[Any, list[Callable]] = {
diff --git a/pwndbg/gdblib/prompt.py b/pwndbg/gdblib/prompt.py
--- a/pwndbg/gdblib/prompt.py
+++ b/pwndbg/gdblib/prompt.py
@@ -89,15 +89,4 @@
         prompt = "\x01" + prompt + "\x02"  # SOH + prompt + STX
 
     gdb.execute(f"set prompt {prompt}")
-
-
-if pwndbg.gdblib.events.before_prompt_event.is_real_event:
-    gdb.prompt_hook = initial_hook
-
-else:
-    # Old GDBs doesn't have gdb.events.before_prompt, so we will emulate it using gdb.prompt_hook
-    def extended_prompt_hook(*a):
-        pwndbg.gdblib.events.before_prompt_event.invoke_callbacks()
-        return prompt_hook(*a)
-
-    gdb.prompt_hook = extended_prompt_hook
+gdb.prompt_hook = initial_hook

{"golden_diff": "diff --git a/pwndbg/gdblib/events.py b/pwndbg/gdblib/events.py\n--- a/pwndbg/gdblib/events.py\n+++ b/pwndbg/gdblib/events.py\n@@ -67,44 +67,6 @@\n \n gdb.events.start = StartEvent()\n \n-\n-class EventWrapper:\n-    \"\"\"\n-    Wrapper for GDB events which may not exist on older GDB versions but we still can\n-    fire them manually (to invoke them you have to call `invoke_callbacks`).\n-    \"\"\"\n-\n-    def __init__(self, name: str) -> None:\n-        self.name = name\n-\n-        self._event = getattr(gdb.events, self.name, None)\n-        self._is_real_event = self._event is not None\n-\n-    def connect(self, func) -> None:\n-        if self._event is not None:\n-            self._event.connect(func)\n-\n-    def disconnect(self, func) -> None:\n-        if self._event is not None:\n-            self._event.disconnect(func)\n-\n-    @property\n-    def is_real_event(self) -> bool:\n-        return self._is_real_event\n-\n-    def invoke_callbacks(self) -> None:\n-        \"\"\"\n-        As an optimization please don't call this if your GDB has this event (check `.is_real_event`).\n-        \"\"\"\n-        for f in registered[self]:\n-            f()\n-\n-\n-# Old GDBs doesn't have gdb.events.before_prompt, so we will emulate it using gdb.prompt_hook\n-before_prompt_event = EventWrapper(\"before_prompt\")\n-gdb.events.before_prompt = before_prompt_event\n-\n-\n # In order to support reloading, we must be able to re-fire\n # all 'objfile' and 'stop' events.\n registered: dict[Any, list[Callable]] = {\ndiff --git a/pwndbg/gdblib/prompt.py b/pwndbg/gdblib/prompt.py\n--- a/pwndbg/gdblib/prompt.py\n+++ b/pwndbg/gdblib/prompt.py\n@@ -89,15 +89,4 @@\n         prompt = \"\\x01\" + prompt + \"\\x02\"  # SOH + prompt + STX\n \n     gdb.execute(f\"set prompt {prompt}\")\n-\n-\n-if pwndbg.gdblib.events.before_prompt_event.is_real_event:\n-    gdb.prompt_hook = initial_hook\n-\n-else:\n-    # Old GDBs doesn't have gdb.events.before_prompt, so we will emulate it using gdb.prompt_hook\n-    def extended_prompt_hook(*a):\n-        pwndbg.gdblib.events.before_prompt_event.invoke_callbacks()\n-        return prompt_hook(*a)\n-\n-    gdb.prompt_hook = extended_prompt_hook\n+gdb.prompt_hook = initial_hook\n", "issue": "Refactor EventWrapper\nIt seems we can maybe remove EventWrapper class now since we depend on GDB >= 9.2?\r\n\r\nhttps://github.com/pwndbg/pwndbg/issues/1854#issuecomment-1664527390\n", "before_files": [{"content": "\"\"\"\nEnables callbacks into functions to be automatically invoked\nwhen various events occur to the debuggee (e.g. STOP on SIGINT)\nby using a decorator.\n\"\"\"\n\nfrom __future__ import annotations\n\nimport sys\nfrom functools import partial\nfrom functools import wraps\nfrom typing import Any\nfrom typing import Callable\n\nimport gdb\n\nfrom pwndbg.gdblib.config import config\n\ndebug = config.add_param(\"debug-events\", False, \"display internal event debugging info\")\n\n\n# There is no GDB way to get a notification when the binary itself\n# is loaded from disk, by the operating system, before absolutely\n# anything happens\n#\n# However, we get an Objfile event when the binary is loaded, before\n# its entry point is invoked.\n#\n# We also get an Objfile event when we load up GDB, so we need\n# to detect when the binary is running or not.\n#\n# Additionally, when attaching to a process running under QEMU, the\n# very first event which is fired is a 'stop' event.  We need to\n# capture this so that we can fire off all of the 'start' events first.\nclass StartEvent:\n    def __init__(self) -> None:\n        self.registered: list[Callable] = []\n        self.running = False\n\n    def connect(self, function) -> None:\n        if function not in self.registered:\n            self.registered.append(function)\n\n    def disconnect(self, function) -> None:\n        if function in self.registered:\n            self.registered.remove(function)\n\n    def on_new_objfile(self) -> None:\n        if self.running or not gdb.selected_thread():\n            return\n\n        self.running = True\n\n        for function in self.registered:\n            if debug:\n                sys.stdout.write(\n                    \"{!r} {}.{}\\n\".format(\"start\", function.__module__, function.__name__)\n                )\n            function()\n\n    def on_exited(self) -> None:\n        self.running = False\n\n    def on_stop(self) -> None:\n        self.on_new_objfile()\n\n\ngdb.events.start = StartEvent()\n\n\nclass EventWrapper:\n    \"\"\"\n    Wrapper for GDB events which may not exist on older GDB versions but we still can\n    fire them manually (to invoke them you have to call `invoke_callbacks`).\n    \"\"\"\n\n    def __init__(self, name: str) -> None:\n        self.name = name\n\n        self._event = getattr(gdb.events, self.name, None)\n        self._is_real_event = self._event is not None\n\n    def connect(self, func) -> None:\n        if self._event is not None:\n            self._event.connect(func)\n\n    def disconnect(self, func) -> None:\n        if self._event is not None:\n            self._event.disconnect(func)\n\n    @property\n    def is_real_event(self) -> bool:\n        return self._is_real_event\n\n    def invoke_callbacks(self) -> None:\n        \"\"\"\n        As an optimization please don't call this if your GDB has this event (check `.is_real_event`).\n        \"\"\"\n        for f in registered[self]:\n            f()\n\n\n# Old GDBs doesn't have gdb.events.before_prompt, so we will emulate it using gdb.prompt_hook\nbefore_prompt_event = EventWrapper(\"before_prompt\")\ngdb.events.before_prompt = before_prompt_event\n\n\n# In order to support reloading, we must be able to re-fire\n# all 'objfile' and 'stop' events.\nregistered: dict[Any, list[Callable]] = {\n    gdb.events.exited: [],\n    gdb.events.cont: [],\n    gdb.events.new_objfile: [],\n    gdb.events.stop: [],\n    gdb.events.start: [],\n    gdb.events.new_thread: [],\n    gdb.events.before_prompt: [],  # The real event might not exist, but we wrap it\n    gdb.events.memory_changed: [],\n    gdb.events.register_changed: [],\n}\n\n\n# When performing remote debugging, gdbserver is very noisy about which\n# objects are loaded.  This greatly slows down the debugging session.\n# In order to combat this, we keep track of which objfiles have been loaded\n# this session, and only emit objfile events for each *new* file.\nobjfile_cache: dict[str, set[str]] = {}\n\n\ndef connect(func, event_handler, name=\"\"):\n    if debug:\n        print(\"Connecting\", func.__name__, event_handler)\n\n    @wraps(func)\n    def caller(*a):\n        if debug:\n            sys.stdout.write(f\"{name!r} {func.__module__}.{func.__name__} {a!r}\\n\")\n\n        if a and isinstance(a[0], gdb.NewObjFileEvent):\n            objfile = a[0].new_objfile\n            handler = f\"{func.__module__}.{func.__name__}\"\n            path = objfile.filename\n            dispatched = objfile_cache.get(path, set())\n\n            if handler in dispatched:\n                return\n\n            dispatched.add(handler)\n            objfile_cache[path] = dispatched\n\n        try:\n            func()\n        except Exception as e:\n            import pwndbg.exception\n\n            pwndbg.exception.handle()\n            raise e\n\n    registered[event_handler].append(caller)\n    event_handler.connect(caller)\n    return func\n\n\ndef exit(func):\n    return connect(func, gdb.events.exited, \"exit\")\n\n\ndef cont(func):\n    return connect(func, gdb.events.cont, \"cont\")\n\n\ndef new_objfile(func):\n    return connect(func, gdb.events.new_objfile, \"obj\")\n\n\ndef stop(func):\n    return connect(func, gdb.events.stop, \"stop\")\n\n\ndef start(func):\n    return connect(func, gdb.events.start, \"start\")\n\n\ndef thread(func):\n    return connect(func, gdb.events.new_thread, \"thread\")\n\n\nbefore_prompt = partial(connect, event_handler=gdb.events.before_prompt, name=\"before_prompt\")\n\n\ndef reg_changed(func):\n    try:\n        return connect(func, gdb.events.register_changed, \"reg_changed\")\n    except AttributeError:\n        return func\n\n\ndef mem_changed(func):\n    try:\n        return connect(func, gdb.events.memory_changed, \"mem_changed\")\n    except AttributeError:\n        return func\n\n\ndef log_objfiles(ofile=None) -> None:\n    if not (debug and ofile):\n        return\n\n    name = ofile.new_objfile.filename\n\n    print(\"objfile: %r\" % name)\n    gdb.execute(\"info sharedlibrary\")\n\n\ngdb.events.new_objfile.connect(log_objfiles)\n\n\ndef after_reload(start=True) -> None:\n    if gdb.selected_inferior().pid:\n        for f in registered[gdb.events.stop]:\n            f()\n        for f in registered[gdb.events.start]:\n            if start:\n                f()\n        for f in registered[gdb.events.new_objfile]:\n            f()\n        for f in registered[gdb.events.before_prompt]:\n            f()\n\n\ndef on_reload() -> None:\n    for event, functions in registered.items():\n        for function in functions:\n            event.disconnect(function)\n        registered[event] = []\n\n\n@new_objfile\ndef _start_newobjfile() -> None:\n    gdb.events.start.on_new_objfile()\n\n\n@exit\ndef _start_exit() -> None:\n    gdb.events.start.on_exited()\n\n\n@stop\ndef _start_stop() -> None:\n    gdb.events.start.on_stop()\n\n\n@exit\ndef _reset_objfiles() -> None:\n    global objfile_cache\n    objfile_cache = {}\n", "path": "pwndbg/gdblib/events.py"}, {"content": "from __future__ import annotations\n\nfrom os import environ\n\nimport gdb\n\nimport pwndbg.decorators\nimport pwndbg.gdblib.events\nimport pwndbg.gdblib.functions\nimport pwndbg.lib.cache\nimport pwndbg.profiling\nfrom pwndbg.color import disable_colors\nfrom pwndbg.color import message\nfrom pwndbg.lib.tips import color_tip\nfrom pwndbg.lib.tips import get_tip_of_the_day\n\nfuncs_list_str = \", \".join(message.notice(\"$\" + f.name) for f in pwndbg.gdblib.functions.functions)\n\nnum_pwndbg_cmds = sum(\n    1 for _ in filter(lambda c: not (c.shell or c.is_alias), pwndbg.commands.commands)\n)\nnum_shell_cmds = sum(1 for _ in filter(lambda c: c.shell, pwndbg.commands.commands))\nhint_lines = (\n    \"loaded %i pwndbg commands and %i shell commands. Type %s for a list.\"\n    % (num_pwndbg_cmds, num_shell_cmds, message.notice(\"pwndbg [--shell | --all] [filter]\")),\n    f\"created {funcs_list_str} GDB functions (can be used with print/break)\",\n)\n\nfor line in hint_lines:\n    print(message.prompt(\"pwndbg: \") + message.system(line))\n\n# noinspection PyPackageRequirements\nshow_tip = pwndbg.gdblib.config.add_param(\n    \"show-tips\", True, \"whether to display the tip of the day on startup\"\n)\n\ncur = None\n\n\ndef initial_hook(*a) -> None:\n    if show_tip and not pwndbg.decorators.first_prompt:\n        colored_tip = color_tip(get_tip_of_the_day())\n        print(\n            message.prompt(\"------- tip of the day\")\n            + message.system(\" (disable with %s)\" % message.notice(\"set show-tips off\"))\n            + message.prompt(\" -------\")\n        )\n        print(colored_tip)\n    pwndbg.decorators.first_prompt = True\n\n    prompt_hook(*a)\n\n    if environ.get(\"PWNDBG_PROFILE\") == \"1\":\n        pwndbg.profiling.profiler.stop(\"pwndbg-first-prompt.pstats\")\n\n    gdb.prompt_hook = prompt_hook\n\n\ncontext_shown = False\n\n\ndef prompt_hook(*a) -> None:\n    global cur, context_shown\n\n    new = (gdb.selected_inferior(), gdb.selected_thread())\n\n    if cur != new:\n        pwndbg.gdblib.events.after_reload(start=cur is None)\n        cur = new\n\n    if pwndbg.gdblib.proc.alive and pwndbg.gdblib.proc.thread_is_stopped and not context_shown:\n        pwndbg.commands.context.context()\n        context_shown = True\n\n\n@pwndbg.gdblib.events.cont\ndef reset_context_shown(*a) -> None:\n    global context_shown\n    context_shown = False\n\n\n@pwndbg.gdblib.config.trigger(message.config_prompt_color, disable_colors)\ndef set_prompt() -> None:\n    prompt = \"pwndbg> \"\n\n    if not disable_colors:\n        prompt = \"\\x02\" + prompt + \"\\x01\"  # STX + prompt + SOH\n        prompt = message.prompt(prompt)\n        prompt = \"\\x01\" + prompt + \"\\x02\"  # SOH + prompt + STX\n\n    gdb.execute(f\"set prompt {prompt}\")\n\n\nif pwndbg.gdblib.events.before_prompt_event.is_real_event:\n    gdb.prompt_hook = initial_hook\n\nelse:\n    # Old GDBs doesn't have gdb.events.before_prompt, so we will emulate it using gdb.prompt_hook\n    def extended_prompt_hook(*a):\n        pwndbg.gdblib.events.before_prompt_event.invoke_callbacks()\n        return prompt_hook(*a)\n\n    gdb.prompt_hook = extended_prompt_hook\n", "path": "pwndbg/gdblib/prompt.py"}], "after_files": [{"content": "\"\"\"\nEnables callbacks into functions to be automatically invoked\nwhen various events occur to the debuggee (e.g. STOP on SIGINT)\nby using a decorator.\n\"\"\"\n\nfrom __future__ import annotations\n\nimport sys\nfrom functools import partial\nfrom functools import wraps\nfrom typing import Any\nfrom typing import Callable\n\nimport gdb\n\nfrom pwndbg.gdblib.config import config\n\ndebug = config.add_param(\"debug-events\", False, \"display internal event debugging info\")\n\n\n# There is no GDB way to get a notification when the binary itself\n# is loaded from disk, by the operating system, before absolutely\n# anything happens\n#\n# However, we get an Objfile event when the binary is loaded, before\n# its entry point is invoked.\n#\n# We also get an Objfile event when we load up GDB, so we need\n# to detect when the binary is running or not.\n#\n# Additionally, when attaching to a process running under QEMU, the\n# very first event which is fired is a 'stop' event.  We need to\n# capture this so that we can fire off all of the 'start' events first.\nclass StartEvent:\n    def __init__(self) -> None:\n        self.registered: list[Callable] = []\n        self.running = False\n\n    def connect(self, function) -> None:\n        if function not in self.registered:\n            self.registered.append(function)\n\n    def disconnect(self, function) -> None:\n        if function in self.registered:\n            self.registered.remove(function)\n\n    def on_new_objfile(self) -> None:\n        if self.running or not gdb.selected_thread():\n            return\n\n        self.running = True\n\n        for function in self.registered:\n            if debug:\n                sys.stdout.write(\n                    \"{!r} {}.{}\\n\".format(\"start\", function.__module__, function.__name__)\n                )\n            function()\n\n    def on_exited(self) -> None:\n        self.running = False\n\n    def on_stop(self) -> None:\n        self.on_new_objfile()\n\n\ngdb.events.start = StartEvent()\n\n# In order to support reloading, we must be able to re-fire\n# all 'objfile' and 'stop' events.\nregistered: dict[Any, list[Callable]] = {\n    gdb.events.exited: [],\n    gdb.events.cont: [],\n    gdb.events.new_objfile: [],\n    gdb.events.stop: [],\n    gdb.events.start: [],\n    gdb.events.new_thread: [],\n    gdb.events.before_prompt: [],  # The real event might not exist, but we wrap it\n    gdb.events.memory_changed: [],\n    gdb.events.register_changed: [],\n}\n\n\n# When performing remote debugging, gdbserver is very noisy about which\n# objects are loaded.  This greatly slows down the debugging session.\n# In order to combat this, we keep track of which objfiles have been loaded\n# this session, and only emit objfile events for each *new* file.\nobjfile_cache: dict[str, set[str]] = {}\n\n\ndef connect(func, event_handler, name=\"\"):\n    if debug:\n        print(\"Connecting\", func.__name__, event_handler)\n\n    @wraps(func)\n    def caller(*a):\n        if debug:\n            sys.stdout.write(f\"{name!r} {func.__module__}.{func.__name__} {a!r}\\n\")\n\n        if a and isinstance(a[0], gdb.NewObjFileEvent):\n            objfile = a[0].new_objfile\n            handler = f\"{func.__module__}.{func.__name__}\"\n            path = objfile.filename\n            dispatched = objfile_cache.get(path, set())\n\n            if handler in dispatched:\n                return\n\n            dispatched.add(handler)\n            objfile_cache[path] = dispatched\n\n        try:\n            func()\n        except Exception as e:\n            import pwndbg.exception\n\n            pwndbg.exception.handle()\n            raise e\n\n    registered[event_handler].append(caller)\n    event_handler.connect(caller)\n    return func\n\n\ndef exit(func):\n    return connect(func, gdb.events.exited, \"exit\")\n\n\ndef cont(func):\n    return connect(func, gdb.events.cont, \"cont\")\n\n\ndef new_objfile(func):\n    return connect(func, gdb.events.new_objfile, \"obj\")\n\n\ndef stop(func):\n    return connect(func, gdb.events.stop, \"stop\")\n\n\ndef start(func):\n    return connect(func, gdb.events.start, \"start\")\n\n\ndef thread(func):\n    return connect(func, gdb.events.new_thread, \"thread\")\n\n\nbefore_prompt = partial(connect, event_handler=gdb.events.before_prompt, name=\"before_prompt\")\n\n\ndef reg_changed(func):\n    try:\n        return connect(func, gdb.events.register_changed, \"reg_changed\")\n    except AttributeError:\n        return func\n\n\ndef mem_changed(func):\n    try:\n        return connect(func, gdb.events.memory_changed, \"mem_changed\")\n    except AttributeError:\n        return func\n\n\ndef log_objfiles(ofile=None) -> None:\n    if not (debug and ofile):\n        return\n\n    name = ofile.new_objfile.filename\n\n    print(\"objfile: %r\" % name)\n    gdb.execute(\"info sharedlibrary\")\n\n\ngdb.events.new_objfile.connect(log_objfiles)\n\n\ndef after_reload(start=True) -> None:\n    if gdb.selected_inferior().pid:\n        for f in registered[gdb.events.stop]:\n            f()\n        for f in registered[gdb.events.start]:\n            if start:\n                f()\n        for f in registered[gdb.events.new_objfile]:\n            f()\n        for f in registered[gdb.events.before_prompt]:\n            f()\n\n\ndef on_reload() -> None:\n    for event, functions in registered.items():\n        for function in functions:\n            event.disconnect(function)\n        registered[event] = []\n\n\n@new_objfile\ndef _start_newobjfile() -> None:\n    gdb.events.start.on_new_objfile()\n\n\n@exit\ndef _start_exit() -> None:\n    gdb.events.start.on_exited()\n\n\n@stop\ndef _start_stop() -> None:\n    gdb.events.start.on_stop()\n\n\n@exit\ndef _reset_objfiles() -> None:\n    global objfile_cache\n    objfile_cache = {}\n", "path": "pwndbg/gdblib/events.py"}, {"content": "from __future__ import annotations\n\nfrom os import environ\n\nimport gdb\n\nimport pwndbg.decorators\nimport pwndbg.gdblib.events\nimport pwndbg.gdblib.functions\nimport pwndbg.lib.cache\nimport pwndbg.profiling\nfrom pwndbg.color import disable_colors\nfrom pwndbg.color import message\nfrom pwndbg.lib.tips import color_tip\nfrom pwndbg.lib.tips import get_tip_of_the_day\n\nfuncs_list_str = \", \".join(message.notice(\"$\" + f.name) for f in pwndbg.gdblib.functions.functions)\n\nnum_pwndbg_cmds = sum(\n    1 for _ in filter(lambda c: not (c.shell or c.is_alias), pwndbg.commands.commands)\n)\nnum_shell_cmds = sum(1 for _ in filter(lambda c: c.shell, pwndbg.commands.commands))\nhint_lines = (\n    \"loaded %i pwndbg commands and %i shell commands. Type %s for a list.\"\n    % (num_pwndbg_cmds, num_shell_cmds, message.notice(\"pwndbg [--shell | --all] [filter]\")),\n    f\"created {funcs_list_str} GDB functions (can be used with print/break)\",\n)\n\nfor line in hint_lines:\n    print(message.prompt(\"pwndbg: \") + message.system(line))\n\n# noinspection PyPackageRequirements\nshow_tip = pwndbg.gdblib.config.add_param(\n    \"show-tips\", True, \"whether to display the tip of the day on startup\"\n)\n\ncur = None\n\n\ndef initial_hook(*a) -> None:\n    if show_tip and not pwndbg.decorators.first_prompt:\n        colored_tip = color_tip(get_tip_of_the_day())\n        print(\n            message.prompt(\"------- tip of the day\")\n            + message.system(\" (disable with %s)\" % message.notice(\"set show-tips off\"))\n            + message.prompt(\" -------\")\n        )\n        print(colored_tip)\n    pwndbg.decorators.first_prompt = True\n\n    prompt_hook(*a)\n\n    if environ.get(\"PWNDBG_PROFILE\") == \"1\":\n        pwndbg.profiling.profiler.stop(\"pwndbg-first-prompt.pstats\")\n\n    gdb.prompt_hook = prompt_hook\n\n\ncontext_shown = False\n\n\ndef prompt_hook(*a) -> None:\n    global cur, context_shown\n\n    new = (gdb.selected_inferior(), gdb.selected_thread())\n\n    if cur != new:\n        pwndbg.gdblib.events.after_reload(start=cur is None)\n        cur = new\n\n    if pwndbg.gdblib.proc.alive and pwndbg.gdblib.proc.thread_is_stopped and not context_shown:\n        pwndbg.commands.context.context()\n        context_shown = True\n\n\n@pwndbg.gdblib.events.cont\ndef reset_context_shown(*a) -> None:\n    global context_shown\n    context_shown = False\n\n\n@pwndbg.gdblib.config.trigger(message.config_prompt_color, disable_colors)\ndef set_prompt() -> None:\n    prompt = \"pwndbg> \"\n\n    if not disable_colors:\n        prompt = \"\\x02\" + prompt + \"\\x01\"  # STX + prompt + SOH\n        prompt = message.prompt(prompt)\n        prompt = \"\\x01\" + prompt + \"\\x02\"  # SOH + prompt + STX\n\n    gdb.execute(f\"set prompt {prompt}\")\ngdb.prompt_hook = initial_hook\n", "path": "pwndbg/gdblib/prompt.py"}]}