Working on getting interactive working

pwncat/channel/__init__.py

@@ -9,10 +9,29 @@ CHANNEL_TYPES = {}
 class ChannelError(Exception):
     """ Raised when a channel fails to connect """
 
+    def __init__(self, ch, msg="generic channel failure"):
+        super().__init__(msg)
+        self.channel = ch
+
 
 class ChannelClosed(ChannelError):
     """ A channel was closed unexpectedly during communication """
 
+    def __init__(self, ch):
+        super().__init__(ch, "channel unexpectedly closed")
+
+    def cleanup(self, manager: "pwncat.manager.Manager"):
+        """ Cleanup this channel from the manager """
+
+        # If we don't have a session, there's nothing to do
+        session = manager.find_session_by_channel(self.channel)
+        if session is None:
+            return
+
+        # Session takes care of removing itself from the manager
+        # and unsetting `manager.target` if needed.
+        session.died()
+
 
 class ChannelTimeout(ChannelError):
     """ Raised when a read times out.
@@ -21,8 +40,8 @@ class ChannelTimeout(ChannelError):
     :type data: bytes
     """
 
-    def __init__(self, data: bytes):
-        super().__init__("channel recieve timed out")
+    def __init__(self, ch, data: bytes):
+        super().__init__(ch, "channel recieve timed out")
         self.data: bytes = data
 
 
@@ -185,6 +204,25 @@ class Channel:
 
         self.peek_buffer: bytes = b""
 
+    @property
+    def connected(self):
+        """ Check if this channel is connected. This should return
+        false prior to an established connection, and may return
+        true prior to the ``connect`` method being called for some
+        channel types. """
+
+    def connect(self):
+        """ Utilize the parameters provided at initialization to
+        connect to the remote host. This is mainly used for channels
+        which listen for a connection. In that case, `__init__` creates
+        the listener while connect actually establishes a connection.
+        For direct connection-type channels, all logic can be implemented
+        in the constructor.
+
+        This method is called when creating a platform around this channel
+        to instantiate the session.
+        """
+
     def send(self, data: bytes):
         """ Send data to the remote shell. This is a blocking call
         that only returns after all data is sent. """
@@ -353,6 +391,15 @@ class Channel:
         else:
             return BufferedWriter(raw_io, buffer_size=bufsize)
 
+    def close(self):
+        """ Close this channel. This method should do nothing if
+        the ``connected`` property returns False. """
+
+    def __str__(self):
+        """ Get a representation of this channel """
+
+        return f"[cyan]{self.address[0]}[/cyan]:[blue]{self.address[1]}[/blue]"
+
 
 def register(name: str, channel_class):
     """

pwncat/commands/__init__.py

@@ -41,7 +41,7 @@ import pwncat
 import pwncat.db
 from pwncat.commands.base import CommandDefinition, Complete
 from pwncat.util import State, console
-from pwncat.db import get_session
+from pwncat.channel import ChannelClosed
 
 
 def resolve_blocks(source: str):
@@ -104,20 +104,27 @@ def resolve_blocks(source: str):
 class DatabaseHistory(History):
     """ Yield history from the host entry in the database """
 
+    def __init__(self, manager):
+        super().__init__()
+        self.manager = manager
+
     def load_history_strings(self) -> Iterable[str]:
         """ Load the history from the database """
-        for history in (
-            get_session()
-            .query(pwncat.db.History)
-            .order_by(pwncat.db.History.id.desc())
-            .all()
-        ):
-            yield history.command
+
+        with self.manager.new_db_session() as session:
+            for history in (
+                session.query(pwncat.db.History)
+                .order_by(pwncat.db.History.id.desc())
+                .all()
+            ):
+                yield history.command
 
     def store_string(self, string: str) -> None:
         """ Store a command in the database """
-        history = pwncat.db.History(host_id=pwncat.victim.host.id, command=string)
-        get_session().add(history)
+        history = pwncat.db.History(command=string)
+
+        with self.manager.new_db_session() as session:
+            session.add(history)
 
 
 class CommandParser:
@@ -127,16 +134,17 @@ class CommandParser:
     termios modes for the control tty at will in order to support raw vs
     command mode. """
 
-    def __init__(self):
+    def __init__(self, manager: "pwncat.manager.Manager"):
         """ We need to dynamically load commands from pwncat.commands """
 
+        self.manager = manager
         self.commands: List["CommandDefinition"] = []
 
         for loader, module_name, is_pkg in pkgutil.walk_packages(__path__):
             if module_name == "base":
                 continue
             self.commands.append(
-                loader.find_module(module_name).load_module(module_name).Command()
+                loader.find_module(module_name).load_module(module_name).Command(self)
             )
 
         self.prompt: PromptSession = None
@@ -153,12 +161,8 @@ class CommandParser:
         """ This needs to happen after __init__ when the database is fully
         initialized. """
 
-        if pwncat.victim is not None and pwncat.victim.host is not None:
-            history = DatabaseHistory()
-        else:
-            history = InMemoryHistory()
-
-        completer = CommandCompleter(self.commands)
+        history = DatabaseHistory(self.manager)
+        completer = CommandCompleter(self.manager, self.commands)
         lexer = PygmentsLexer(CommandLexer.build(self.commands))
         style = style_from_pygments_cls(get_style_by_name("monokai"))
         auto_suggest = AutoSuggestFromHistory()
@@ -177,28 +181,20 @@ class CommandParser:
             history=history,
         )
 
-    @property
-    def loaded(self):
-        return self.loading_complete
-
-    @loaded.setter
-    def loaded(self, value: bool):
-        assert value == True
-        self.loading_complete = True
-        self.eval(pwncat.config["on_load"], "on_load")
-
     def eval(self, source: str, name: str = "<script>"):
         """ Evaluate the given source file. This will execute the given string
         as a script of commands. Syntax is the same except that commands may
         be separated by semicolons, comments are accepted as following a "#" and
         multiline strings are supported with '"{' and '}"' as delimeters. """
 
-        in_multiline_string = False
-        lineno = 1
-
         for command in resolve_blocks(source):
             try:
                 self.dispatch_line(command)
+            except ChannelClosed as exc:
+                # A channel was unexpectedly closed
+                self.manager.log(f"[red]warning[/red]: {exc.channel}: channel closed")
+                # Ensure any existing sessions are cleaned from the manager
+                exc.cleanup(self.manager)
             except Exception as exc:
                 console.log(
                     f"[red]error[/red]: [cyan]{name}[/cyan]: [yellow]{command}[/yellow]: {str(exc)}"
@@ -207,6 +203,9 @@ class CommandParser:
 
     def run_single(self):
 
+        if self.prompt is None:
+            self.setup_prompt()
+
         try:
             line = self.prompt.prompt().strip()
         except (EOFError, OSError, KeyboardInterrupt):
@@ -217,14 +216,20 @@ class CommandParser:
 
     def run(self):
 
-        self.running = True
+        if self.prompt is None:
+            self.setup_prompt()
 
-        while self.running:
+        running = True
+
+        while running:
             try:
 
-                if pwncat.config.module:
+                if self.manager.config.module:
                     self.prompt.message = [
-                        ("fg:ansiyellow bold", f"({pwncat.config.module.name}) ",),
+                        (
+                            "fg:ansiyellow bold",
+                            f"({self.manager.config.module.name}) ",
+                        ),
                         ("fg:ansimagenta bold", "pwncat"),
                         ("", "$ "),
                     ]
@@ -245,22 +250,20 @@ class CommandParser:
             # badly written command from completely killing our remote
             # connection.
             except EOFError:
-                # We don't have a connection yet, just exit
-                if pwncat.victim is None or pwncat.victim.client is None:
-                    break
-                # We have a connection! Go back to raw mode
-                pwncat.victim.state = State.RAW
-                self.running = False
+                # C-d was pressed. Assume we want to exit the prompt.
+                running = False
             except KeyboardInterrupt:
+                # Normal C-c from a shell just clears the current prompt
                 continue
+            except ChannelClosed as exc:
+                # A channel was unexpectedly closed
+                self.manager.log(f"[red]warning[/red]: {exc.channel}: channel closed")
+                # Ensure any existing sessions are cleaned from the manager
+                exc.cleanup(self.manager)
             except (Exception, KeyboardInterrupt):
                 console.print_exception(width=None)
                 continue
 
-    #             except KeyboardInterrupt:
-    #                 console.log("Keyboard Interrupt")
-    #                 continue
-
     def dispatch_line(self, line: str, prog_name: str = None):
         """ Parse the given line of command input and dispatch a command """
 
@@ -273,7 +276,7 @@ class CommandParser:
             # Spit the line with shell rules
             argv = shlex.split(line)
         except ValueError as e:
-            console.log(f"[red]error[/red]: {e.args[0]}")
+            self.manager.log(f"[red]error[/red]: {e.args[0]}")
             return
 
         if argv[0][0] in self.shortcuts:
@@ -291,12 +294,12 @@ class CommandParser:
                 if argv[0] in self.aliases:
                     command = self.aliases[argv[0]]
                 else:
-                    console.log(f"[red]error[/red]: {argv[0]}: unknown command")
+                    self.manager.log(f"[red]error[/red]: {argv[0]}: unknown command")
                     return
 
-            if not self.loading_complete and not command.LOCAL:
-                console.log(
-                    f"[red]error[/red]: {argv[0]}: non-local command use before connection"
+            if self.manager.target is None and not command.LOCAL:
+                self.manager.log(
+                    f"[red]error[/red]: {argv[0]}: active session required"
                 )
                 return
 
@@ -317,7 +320,7 @@ class CommandParser:
                 args = line
 
             # Run the command
-            command.run(args)
+            command.run(self.manager, args)
 
             if prog_name:
                 command.parser.prog = prog_name
@@ -481,19 +484,22 @@ class CommandLexer(RegexLexer):
 class RemotePathCompleter(Completer):
     """ Complete remote file names/paths """
 
+    def __init__(self, manager: "pwncat.manager.Manager", *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.manager = manager
+
     def get_completions(self, document: Document, complete_event: CompleteEvent):
 
+        if self.manager.target is None:
+            return
+
         before = document.text_before_cursor.split()[-1]
         path, partial_name = os.path.split(before)
 
         if path == "":
             path = "."
 
-        pipe = pwncat.victim.subprocess(
-            f"ls -1 -a --color=never {shlex.quote(path)}", "r"
-        )
-
-        for name in pipe:
+        for name in self.manager.target.listdir(path):
             name = name.decode("utf-8").strip()
             if name.startswith(partial_name):
                 yield Completion(
@@ -530,12 +536,14 @@ class LocalPathCompleter(Completer):
 class CommandCompleter(Completer):
     """ Complete commands from a given list of commands """
 
-    def __init__(self, commands: List["CommandDefinition"]):
+    def __init__(
+        self, manager: "pwncat.manager.Manager", commands: List["CommandDefinition"]
+    ):
         """ Construct a new command completer """
 
         self.layers = {}
         local_file_completer = LocalPathCompleter()
-        remote_file_completer = RemotePathCompleter()
+        remote_file_completer = RemotePathCompleter(manager)
 
         for command in commands:
             self.layers[command.PROG] = [None, [], {}]

pwncat/commands/base.py

@@ -211,10 +211,12 @@ class CommandDefinition:
     #     ),
     # }
 
-    def __init__(self):
+    def __init__(self, manager: "pwncat.manager.Manager"):
         """ Initialize a new command instance. Parse the local arguments array
         into an argparse object. """
 
+        self.manager = manager
+
         # Create the parser object
         if self.ARGS is not None:
             self.parser = argparse.ArgumentParser(
@@ -226,11 +228,13 @@ class CommandDefinition:
         else:
             self.parser = None
 
-    def run(self, args):
+    def run(self, manager: "pwncat.manager.Manager", args):
         """
         This is the "main" for your new command. This should perform the action
         represented by your command.
-        
+
+        :param manager: the manager to operate on
+        :type manager: pwncat.manager.Manager
         :param args: the argparse Namespace containing your parsed arguments
         """
         raise NotImplementedError

pwncat/commands/set.py

@@ -13,10 +13,14 @@ class Command(CommandDefinition):
     """ Set variable runtime variable parameters for pwncat """
 
     def get_config_variables(self):
-        options = ["state"] + list(pwncat.config.values) + list(pwncat.victim.users)
+        options = (
+            ["state"]
+            + list(self.manager.config.values)
+            + list(self.manager.victim.users)
+        )
 
         if pwncat.config.module:
-            options.extend(pwncat.config.module.ARGUMENTS.keys())
+            options.extend(self.manager.config.module.ARGUMENTS.keys())
 
         return options
 
@@ -44,7 +48,7 @@ class Command(CommandDefinition):
     }
     LOCAL = True
 
-    def run(self, args):
+    def run(self, manager, args):
         if args.password:
             if args.variable is None:
                 found = False
@@ -64,18 +68,11 @@ class Command(CommandDefinition):
                 )
                 pwncat.victim.users[args.variable].password = args.value
         else:
-            if (
-                args.variable is not None
-                and args.variable == "state"
-                and args.value is not None
-            ):
+            if args.variable is not None and args.value is not None:
                 try:
-                    pwncat.victim.state = State._member_map_[args.value.upper()]
-                except KeyError:
-                    console.log(f"[red]error[/red]: {args.value}: invalid state")
-            elif args.variable is not None and args.value is not None:
-                try:
-                    pwncat.config.set(
+                    if manager.sessions and args.variable == "db":
+                        raise ValueError("cannot change database with running session")
+                    manager.config.set(
                         args.variable, args.value, getattr(args, "global")
                     )
                     if args.variable == "db":

pwncat/config.py

@@ -104,6 +104,14 @@ class Config:
 
         self.locals[name] = self.module.ARGUMENTS[name].type(value)
 
+    def get(self, name: str, default=None):
+        """ get a value """
+
+        try:
+            return self[name]
+        except KeyError:
+            return default
+
     def use(self, module: BaseModule):
         """ Use the specified module. This clears the current
         locals configuration. """

pwncat/manager.py

@@ -11,8 +11,11 @@ import rich.progress
 
 import pwncat.db
 import pwncat.modules
+from pwncat.util import console
 from pwncat.platform import Platform
 from pwncat.channel import Channel
+from pwncat.config import Config
+from pwncat.commands import CommandParser
 
 
 class Session:
@@ -24,30 +27,36 @@ class Session:
         manager,
         platform: Union[str, Platform],
         channel: Optional[Channel] = None,
-        **kwargs
+        **kwargs,
     ):
         self.manager = manager
+        self.background = None
+        self._db_session = None
+
+        # If necessary, build a new platform object
+        if isinstance(platform, Platform):
+            self.platform = platform
+        else:
+            # If necessary, build a new channel
+            if channel is None:
+                channel = pwncat.channel.create(**kwargs)
+
+            self.platform = pwncat.platform.find(platform)(
+                self, channel, self.config.get("log", None)
+            )
+
         self._progress = rich.progress.Progress(
-            str(platform),
+            str(self.platform),
             "•",
             "{task.description}",
             "•",
             "{task.fields[status]}",
             transient=True,
         )
-        self.config = {}
-        self.background = None
-        self._db_session = None
 
-        if isinstance(platform, Platform):
-            self.platform = platform
-        else:
-            if channel is None:
-                channel = pwncat.channel.create(**kwargs)
-
-            self.platform = pwncat.platform.find(platform)(
-                self, channel, self.get("log")
-            )
+        # Register this session with the manager
+        self.manager.sessions.append(self)
+        self.manager.target = self
 
         # Initialize the host reference
         self.hash = self.platform.get_host_hash()
@@ -55,6 +64,15 @@ class Session:
             self.host = session.query(pwncat.db.Host).filter_by(hash=self.hash).first()
         if self.host is None:
             self.register_new_host()
+        else:
+            self.log("loaded known host from db")
+
+    @property
+    def config(self):
+        """ Get the configuration object for this manager. This
+        is simply a wrapper for session.manager.config to make
+        accessing configuration a little easier. """
+        return self.manager.config
 
     def register_new_host(self):
         """ Register a new host in the database. This assumes the
@@ -66,18 +84,7 @@ class Session:
         with self.db as session:
             session.add(self.host)
 
-    def get(self, name, default=None):
-        """ Get the value of a configuration item """
-
-        if name not in self.config:
-            return self.manager.get(name, default)
-
-        return self.config[name]
-
-    def set(self, name, value):
-        """ Set the value of a configuration item """
-
-        self.config[name] = value
+        self.log("registered new host w/ db")
 
     def run(self, module: str, **kwargs):
         """ Run a module on this session """
@@ -115,7 +122,7 @@ class Session:
         progress instance to log without messing up progress output
         from other sessions, if we aren't active. """
 
-        self.manager.target.progress.log(*args, **kwargs)
+        self.manager.log(f"{self.platform}:", *args, **kwargs)
 
     @property
     @contextlib.contextmanager
@@ -133,7 +140,7 @@ class Session:
                 self._db_session = self.manager.create_db_session()
             yield self._db_session
         finally:
-            if new_session:
+            if new_session and self._db_session is not None:
                 session = self._db_session
                 self._db_session = None
                 session.close()
@@ -182,13 +189,14 @@ class Manager:
     sessions, and executing modules.
     """
 
-    def __init__(self, config: str):
-        self.config = {}
+    def __init__(self, config: str = "./pwncatrc"):
+        self.config = Config()
         self.sessions: List[Session] = []
         self.modules: Dict[str, pwncat.modules.BaseModule] = {}
         self.engine = None
         self.SessionBuilder = None
         self._target = None
+        self.parser = CommandParser(self)
 
         # Load standard modules
         self.load_modules(*pwncat.modules.__path__)
@@ -208,32 +216,60 @@ class Manager:
         if os.path.isdir(modules_dir):
             self.load_modules(modules_dir)
 
+        # Load global configuration script, if available
+        try:
+            with open("/etc/pwncat/pwncatrc") as filp:
+                self.parser.eval(filp.read(), "/etc/pwncat/pwncatrc")
+        except (FileNotFoundError, PermissionError):
+            pass
+
+        # Load user configuration script
+        user_rc = os.path.join(data_home, "pwncatrc")
+        try:
+            with open(user_rc) as filp:
+                self.parser.eval(filp.read(), user_rc)
+        except (FileNotFoundError, PermissionError):
+            pass
+
+        # Load local configuration script
+        try:
+            with open(config) as filp:
+                self.parser.eval(filp.read(), config)
+        except (FileNotFoundError, PermissionError):
+            pass
+
+    def open_database(self):
+        """ Create the internal engine and session builder
+        for this manager based on the configured database """
+
+        if self.sessions and self.engine is not None:
+            raise RuntimeError("cannot change database after sessions are established")
+
+        self.engine = create_engine(self.config["db"])
+        pwncat.db.Base.metadata.create_all(self.engine)
+        self.SessionBuilder = sessionmaker(bind=self.engine)
+
     def create_db_session(self):
         """ Create a new SQLAlchemy database session and return it """
 
         # Initialize a fallback database if needed
         if self.engine is None:
-            self.set("db", "sqlite:///:memory:")
+            self.config.set("db", "sqlite:///:memory:", glob=True)
+            self.open_database()
 
         return self.SessionBuilder()
 
-    def set(self, key, value):
-        """ Set a configuration item in the global manager """
+    @contextlib.contextmanager
+    def new_db_session(self):
+        """ Track a database session in a context manager """
 
-        self.config[key] = value
+        session = None
 
-        if key == "db":
-            # This is dangerous for background modules
-            if self.engine is not None:
-                self.engine.dispose()
-            self.engine = create_engine(value)
-            pwncat.db.Base.metadata.create_all(self.engine)
-            self.SessionBuilder = sessionmaker(bind=self.engine)
-
-    def get(self, key, default=None):
-        """ Retrieve the value of a configuration item """
-
-        return self.config.get(key, default)
+        try:
+            session = self.create_db_session()
+            yield session
+        finally:
+            pass
 
     def load_modules(self, *paths):
         """ Dynamically load modules from the specified paths
@@ -257,6 +293,14 @@ class Manager:
             # Store it's name so we know it later
             setattr(self.modules[module_name], "name", module_name)
 
+    def log(self, *args, **kwargs):
+        """ Output a log entry """
+
+        if self.target is not None:
+            self.target._progress.log(*args, **kwargs)
+        else:
+            console.log(*args, **kwargs)
+
     @property
     def target(self) -> Session:
         """ Retrieve the currently focused target """
@@ -268,15 +312,14 @@ class Manager:
             raise ValueError("invalid target")
         self._target = value
 
-    def run(self, module: str, **kwargs):
-        """ Execute a module on the currently active target """
-
-    def find_module(self, pattern: str):
-        """ Enumerate modules applicable to the current target """
-
     def interactive(self):
         """ Start interactive prompt """
 
+        # This needs to be a full main loop with raw-mode support
+        # eventually, but I want to get the command parser working for
+        # now. The raw mode is the easy part.
+        self.parser.run()
+
     def create_session(self, platform: str, channel: Channel = None, **kwargs):
         """
         Open a new session from a new or existing platform. If the platform
@@ -289,8 +332,4 @@ class Manager:
         """
 
         session = Session(self, platform, channel, **kwargs)
-        self.sessions.append(session)
-
-        self.target = session
-
         return session

pwncat/platform/__init__.py

@@ -44,6 +44,11 @@ class Platform:
         channel: "pwncat.channel.Channel",
         log: str = None,
     ):
+
+        # This will throw a ChannelError if we can't complete the
+        # connection, so we do it first.
+        channel.connect()
+
         self.session = session
         self.channel = channel
         self.logger = logging.getLogger(str(channel))
@@ -58,6 +63,9 @@ class Platform:
             handler.setFormatter(logging.Formatter("%(asctime)s - %(message)s"))
             self.logger.addHandler(handler)
 
+    def __str__(self):
+        return str(self.channel)
+
     def listdir(self, path=None) -> Generator[str, None, None]:
         """ List the contents of a directory. If ``path`` is None,
         then the contents of the current directory is listed. The

pwncat/platform/linux.py

@@ -562,12 +562,21 @@ class Linux(Platform):
             pkg_resources.resource_filename("pwncat", "data/gtfobins.json"), self.which
         )
 
+        # Ensure history is disabled
+        self.disable_history()
+
         p = self.Popen("[ -t 1 ]")
         if p.wait() == 0:
             self.has_pty = True
         else:
             self.has_pty = False
 
+    def disable_history(self):
+        """ Disable shell history """
+
+        # Ensure history is not tracked
+        self.run("unset HISTFILE; export HISTCONTROL=ignorespace; unset PROMPT_COMMAND")
+
     def get_pty(self):
         """ Spawn a PTY in the current shell. If a PTY is already running
         then this method does nothing. """
@@ -608,6 +617,10 @@ class Linux(Platform):
             if not self.interactive:
                 self._interactive = True
                 self.interactive = False
+
+            # When starting a pty, history is sometimes re-enabled
+            self.disable_history()
+
             return
 
         raise PlatformError("no avialable pty methods")
@@ -623,44 +636,58 @@ class Linux(Platform):
         :rtype: str
         """
 
-        try:
-            result = self.run(
-                "hostname -f", shell=True, check=True, text=True, encoding="utf-8"
-            )
-            hostname = result.stdout.strip()
-        except CalledProcessError:
-            hostname = self.channel.getpeername()[0]
-
-        try:
-            result = self.run(
-                "ifconfig -a", shell=True, check=True, text=True, encoding="utf-8"
-            )
-            ifconfig = result.stdout.strip().lower()
-
-            for line in ifconfig.split("\n"):
-                if "hwaddr" in line and "00:00:00:00:00:00" not in line:
-                    mac = line.split("hwaddr ")[1].split("\n")[0].strip()
-                    break
-                if "ether " in line and "00:00:00:00:00:00" not in line:
-                    mac = line.split("ether ")[1].split(" ")[0]
-                    break
-            else:
-                mac = None
-        except CalledProcessError:
-            # Attempt to use the `ip` command instead
+        with self.session.task("calculating host hash") as task:
             try:
-                result = self.run(
-                    "ip link show", shell=True, check=True, text=True, encoding="utf-8"
+                self.session.update_task(
+                    task, status="retrieving hostname (hostname -f)"
                 )
-                ip_out = result.stdout.strip().lower()
-                for line in ip_out.split("\n"):
-                    if "link/ether" in line and "00:00:00:00:00:00" not in line:
-                        mac = line.split("link/ether ")[1].split(" ")[0]
+                result = self.run(
+                    "hostname -f", shell=True, check=True, text=True, encoding="utf-8"
+                )
+                hostname = result.stdout.strip()
+            except CalledProcessError:
+                hostname = self.channel.getpeername()[0]
+
+            try:
+                self.session.update_task(
+                    task, status="retrieving mac addresses (ifconfig)"
+                )
+                result = self.run(
+                    "ifconfig -a", shell=True, check=True, text=True, encoding="utf-8"
+                )
+                ifconfig = result.stdout.strip().lower()
+
+                for line in ifconfig.split("\n"):
+                    if "hwaddr" in line and "00:00:00:00:00:00" not in line:
+                        mac = line.split("hwaddr ")[1].split("\n")[0].strip()
+                        break
+                    if "ether " in line and "00:00:00:00:00:00" not in line:
+                        mac = line.split("ether ")[1].split(" ")[0]
                         break
                 else:
                     mac = None
             except CalledProcessError:
-                mac = None
+                # Attempt to use the `ip` command instead
+                try:
+                    self.session.update_task(
+                        task, status="retrieving mac addresses (ip link show)"
+                    )
+                    result = self.run(
+                        "ip link show",
+                        shell=True,
+                        check=True,
+                        text=True,
+                        encoding="utf-8",
+                    )
+                    ip_out = result.stdout.strip().lower()
+                    for line in ip_out.split("\n"):
+                        if "link/ether" in line and "00:00:00:00:00:00" not in line:
+                            mac = line.split("link/ether ")[1].split(" ")[0]
+                            break
+                    else:
+                        mac = None
+                except CalledProcessError:
+                    mac = None
 
         # In some (unlikely) cases, `mac` may be None, so we use `str` here.
         identifier = hostname + str(mac)