Agent: Change typehints of agent destination path to PurePath

monkey/infection_monkey/exploit/mssqlexec.py

@@ -1,6 +1,6 @@
 import logging
 import os
-from pathlib import Path
+from pathlib import PurePath
 from time import sleep
 
 import pymssql
@@ -132,7 +132,7 @@ class MSSQLExploiter(HostExploiter):
             raise Exception("Couldn't execute MSSQL exploiter because payload was too long")
         self.run_mssql_commands(array_of_commands)
 
-    def run_monkey(self, monkey_path_on_victim: Path):
+    def run_monkey(self, monkey_path_on_victim: PurePath):
         monkey_launch_command = self.get_monkey_launch_command(monkey_path_on_victim)
         self.run_mssql_command(monkey_launch_command)
         self.run_payload_file()
@@ -142,7 +142,7 @@ class MSSQLExploiter(HostExploiter):
             self.cursor.execute(cmd)
             sleep(MSSQLExploiter.QUERY_BUFFER)
 
-    def upload_monkey(self, monkey_path_on_victim: Path):
+    def upload_monkey(self, monkey_path_on_victim: PurePath):
         monkey_download_command = self.write_download_command_to_payload(monkey_path_on_victim)
         self.run_payload_file()
         self.add_executed_cmd(monkey_download_command.command)
@@ -158,7 +158,7 @@ class MSSQLExploiter(HostExploiter):
         )
         self.run_mssql_command(tmp_dir_removal_command)
 
-    def start_monkey_server(self, monkey_path_on_victim: Path) -> LockedHTTPServer:
+    def start_monkey_server(self, monkey_path_on_victim: PurePath) -> LockedHTTPServer:
         self.agent_http_path, http_thread = HTTPTools.create_locked_transfer(
             self.host, str(monkey_path_on_victim), self.agent_repository
         )
@@ -169,12 +169,12 @@ class MSSQLExploiter(HostExploiter):
         http_thread.stop()
         http_thread.join(LONG_REQUEST_TIMEOUT)
 
-    def write_download_command_to_payload(self, monkey_path_on_victim: Path):
+    def write_download_command_to_payload(self, monkey_path_on_victim: PurePath):
         monkey_download_command = self.get_monkey_download_command(monkey_path_on_victim)
         self.run_mssql_command(monkey_download_command)
         return monkey_download_command
 
-    def get_monkey_launch_command(self, monkey_path_on_victim: Path):
+    def get_monkey_launch_command(self, monkey_path_on_victim: PurePath):
         # Form monkey's launch command
         monkey_args = build_monkey_commandline(
             self.host, self.current_depth - 1, monkey_path_on_victim
@@ -187,7 +187,7 @@ class MSSQLExploiter(HostExploiter):
             suffix=suffix,
         )
 
-    def get_monkey_download_command(self, monkey_path_on_victim: Path):
+    def get_monkey_download_command(self, monkey_path_on_victim: PurePath):
         monkey_download_command = MSSQLExploiter.MONKEY_DOWNLOAD_COMMAND.format(
             http_path=self.agent_http_path, dst_path=str(monkey_path_on_victim)
         )

monkey/infection_monkey/exploit/powershell.py

@@ -1,5 +1,5 @@
 import logging
-from pathlib import Path
+from pathlib import Path, PurePath
 from typing import List, Optional
 
 from infection_monkey.exploit.HostExploiter import HostExploiter
@@ -182,7 +182,7 @@ class PowerShellExploiter(HostExploiter):
                 f"Failed to execute the agent binary on the victim: {ex}"
             )
 
-    def _copy_monkey_binary_to_victim(self, monkey_path_on_victim: Path):
+    def _copy_monkey_binary_to_victim(self, monkey_path_on_victim: PurePath):
 
         temp_monkey_binary_filepath = Path(f"./monkey_temp_bin_{get_random_file_suffix()}")
 

monkey/infection_monkey/exploit/powershell_utils/powershell_client.py

@@ -1,6 +1,6 @@
 import abc
 import logging
-from pathlib import Path
+from pathlib import Path, PurePath
 from typing import Optional
 
 import pypsrp
@@ -64,7 +64,7 @@ class IPowerShellClient(Protocol, metaclass=abc.ABCMeta):
         pass
 
     @abc.abstractmethod
-    def copy_file(self, src: Path, dest: Path) -> bool:
+    def copy_file(self, src: Path, dest: PurePath) -> bool:
         pass
 
     @abc.abstractmethod
@@ -102,7 +102,7 @@ class PowerShellClient(IPowerShellClient):
         output, _, _ = self._client.execute_cmd(cmd)
         return output
 
-    def copy_file(self, src: Path, dest: Path):
+    def copy_file(self, src: Path, dest: PurePath):
         try:
             self._client.copy(str(src), str(dest))
             logger.debug(f"Successfully copied {src} to {dest} on {self._ip_addr}")

monkey/infection_monkey/exploit/sshexec.py

@@ -1,6 +1,6 @@
 import io
 import logging
-from pathlib import Path
+from pathlib import PurePath
 
 import paramiko
 
@@ -265,7 +265,7 @@ class SSHExploiter(HostExploiter):
             return self.exploit_result
 
     def _set_executable_bit_on_agent_binary(
-        self, ftp: paramiko.sftp_client.SFTPClient, monkey_path_on_victim: Path
+        self, ftp: paramiko.sftp_client.SFTPClient, monkey_path_on_victim: PurePath
     ):
         ftp.chmod(str(monkey_path_on_victim), 0o700)
         self.telemetry_messenger.send_telemetry(

monkey/infection_monkey/exploit/tools/smb_tools.py

@@ -2,7 +2,7 @@ import logging
 import ntpath
 import pprint
 from io import BytesIO
-from pathlib import Path
+from pathlib import PurePath
 from typing import Optional
 
 from impacket.dcerpc.v5 import srvs, transport
@@ -22,7 +22,7 @@ class SmbTools(object):
     def copy_file(
         host,
         agent_file: BytesIO,
-        dst_path: Path,
+        dst_path: PurePath,
         username,
         password,
         lm_hash="",
@@ -104,7 +104,7 @@ class SmbTools(object):
 
             if str(dst_path).lower().startswith(share_path.lower()):
                 high_priority_shares += (
-                    (ntpath.sep + str(dst_path)[len(share_path):], share_info),
+                    (ntpath.sep + str(dst_path)[len(share_path) :], share_info),
                 )
 
             low_priority_shares += ((ntpath.sep + file_name, share_info),)