Fixed escalation shell tracking

Author: calebstewart

pwncat/commands/escalate.py

@@ -3,6 +3,7 @@
 
 import pwncat
 from pwncat.util import console
+from pwncat.manager import Manager
 from pwncat.modules import ModuleFailed
 from pwncat.commands import Complete, Parameter, CommandDefinition
 
@@ -133,7 +134,7 @@ def list_abilities(self, manager, args):
         elif not found:
             console.log("[yellow]warning[/yellow]: no direct escalations found")
 
-    def do_escalate(self, manager: "pwncat.manager.Manager", task, user, args):
+    def do_escalate(self, manager: Manager, task, user, args):
         """Execute escalations until we find one that works"""
 
         attempted = []
@@ -178,7 +179,7 @@ def do_escalate(self, manager: "pwncat.manager.Manager", task, user, args):
 
                     time.sleep(0.1)
 
-                    manager.target.platform.refresh_uid()
+                    manager.target.platform.context_changed()
 
                     # Construct the escalation link
                     link = Link(manager.target, escalation, result)
@@ -199,7 +200,10 @@ def do_escalate(self, manager: "pwncat.manager.Manager", task, user, args):
                         manager.print(f" - {link}")
 
                     return result
-                except ModuleFailed:
+                except ModuleFailed as exc:
+                    manager.target.log(
+                        f"{escalation.title(manager.target)}: failed: {exc}"
+                    )
                     failed.append(escalation)
 
             if not args.recursive:
@@ -219,6 +223,7 @@ def do_escalate(self, manager: "pwncat.manager.Manager", task, user, args):
                     time.sleep(0.1)
 
                     manager.target.platform.refresh_uid()
+
                     link = Link(manager.target, escalation, result)
 
                     if escalation.type == "escalate.replace":
@@ -229,5 +234,8 @@ def do_escalate(self, manager: "pwncat.manager.Manager", task, user, args):
                     chain.append(link)
                     attempted.append(escalation.uid)
                     break
-                except ModuleFailed:
+                except ModuleFailed as exc:
+                    manager.target.log(
+                        f"{escalation.title(manager.target)}: failed: {exc}"
+                    )
                     failed.append(escalation)

pwncat/platform/__init__.py

@@ -570,6 +570,10 @@ def __str__(self):
         """Retrieve a string describing the platform connection"""
         return str(self.channel)
 
+    @abstractmethod
+    def context_changed(self):
+        """Used to notify the platform that the pwncat framework changed shell or user context."""
+
     @abstractmethod
     def exit(self):
         """Exit this session"""

pwncat/platform/linux.py

@@ -1047,11 +1047,11 @@ def compile(
                 with open(source, "rb") as src:
                     with self.tempfile(suffix=".c", mode="wb") as dest:
                         shutil.copyfileobj(src, dest)
-                        real_sources.append(dest.name)
+                        real_sources.append(str(dest.name))
             else:
                 with self.tempfile(mode="w", suffix=".c") as dest:
                     shutil.copyfileobj(source, dest)
-                    real_sources.append(dest.name)
+                    real_sources.append(str(dest.name))
 
         if output is None:
             # We just need to create a file...
@@ -1630,6 +1630,24 @@ def sudo(
 
         return proc
 
+    def context_changed(self):
+        """Shell or user context has been manually changed by the framework"""
+
+        # Update self.shell just in case the user changed shells
+        try:
+            # Get the PID of the running shell
+            pid = self.getenv("$")
+            # Grab the path to the executable representing the shell
+            self.shell = self.Path("/proc", pid, "exe").readlink()
+        except (FileNotFoundError, PermissionError, OSError):
+            # Fall back to SHELL even though it's not really trustworthy
+            self.shell = self.getenv("SHELL")
+            if self.shell is None or self.shell == "":
+                self.shell = "/bin/sh"
+
+        # Refresh the currently tracked user and group IDs
+        self.refresh_uid()
+
     @property
     def interactive(self) -> bool:
         """
@@ -1656,20 +1674,8 @@ def interactive(self, value: bool):
             self.channel.drain()
             self._interactive = False
 
-            # Update self.shell just in case the user changed shells
-            try:
-                # Get the PID of the running shell
-                pid = self.getenv("$")
-                # Grab the path to the executable representing the shell
-                self.shell = self.Path("/proc", pid, "exe").readlink()
-            except (FileNotFoundError, PermissionError, OSError):
-                # Fall back to SHELL even though it's not really trustworthy
-                self.shell = self.getenv("SHELL")
-                if self.shell is None or self.shell == "":
-                    self.shell = "/bin/sh"
-
-            # Refresh the currently tracked user and group IDs
-            self.refresh_uid()
+            # update current user and shell variable
+            self.context_changed()
         else:
 
             # Going interactive requires a pty