Merge pull request #1367 from Pennyw0rth/neff-fix-open-calls

Use context manager for file operations

nxc/modules/daclread.py

@@ -249,10 +249,10 @@ class NXCModule:
                 context.log.debug("There is a target specified!")
                 if isfile(module_options[option]):
                     try:
-                        target_file = open(module_options[option])  # ruff: ignore[open-file-with-context-handler]
-                        for line in target_file:
-                            context.log.debug(f"Adding target from file: {line}")
-                            self.targets.append((line.strip(), SEARCH_FILTERS[option]))
+                        with open(module_options[option]) as target_file:
+                            for line in target_file:
+                                context.log.debug(f"Adding target from file: {line}")
+                                self.targets.append((line.strip(), SEARCH_FILTERS[option]))
                     except Exception:
                         context.log.fail("The file doesn't exist or cannot be opened.")
                 else:

nxc/modules/handlekatz.py

@@ -126,20 +126,17 @@ class NXCModule:
             except Exception as e:
                 context.log.fail(f"[OPSEC] Error deleting lsass.dmp file on share {self.share}: {e}")
 
-            h_in = open(self.dir_result + machine_name, "rb")  # ruff: ignore[open-file-with-context-handler]
-            h_out = open(self.dir_result + machine_name + ".decode", "wb")  # ruff: ignore[open-file-with-context-handler]
+            with open(self.dir_result + machine_name, "rb") as h_in, open(self.dir_result + machine_name + ".decode", "wb") as h_out:
+                bytes_in = bytearray(h_in.read())
 
-            bytes_in = bytearray(h_in.read())
-            bytes_in_len = len(bytes_in)
+                context.log.display(f"Deobfuscating, this might take a while (size: {len(bytes_in)} bytes)")
 
-            context.log.display(f"Deobfuscating, this might take a while (size: {bytes_in_len} bytes)")
+                chunks = [bytes_in[i: i + 1000000] for i in range(0, len(bytes_in), 1000000)]
+                for chunk in chunks:
+                    for i in range(len(chunk)):
+                        chunk[i] ^= 0x41
 
-            chunks = [bytes_in[i: i + 1000000] for i in range(0, bytes_in_len, 1000000)]
-            for chunk in chunks:
-                for i in range(len(chunk)):
-                    chunk[i] ^= 0x41
-
-                h_out.write(bytes(chunk))
+                    h_out.write(bytes(chunk))
 
             with open(self.dir_result + machine_name + ".decode", "rb") as dump:
                 try:

nxc/protocols/ftp.py

@@ -128,7 +128,8 @@ class ftp(connection):
             # Check if the file exists
             self.conn.size(filename)
             # Attempt to download the file
-            self.conn.retrbinary(f"RETR {filename}", open(downloaded_file, "wb").write)  # ruff: ignore[open-file-with-context-handler]
+            with open(downloaded_file, "wb") as f:
+                self.conn.retrbinary(f"RETR {filename}", f.write)
         except error_perm as error_message:
             self.logger.fail(f"Failed to download the file. Response: ({error_message})")
             self.conn.close()
@@ -146,7 +147,8 @@ class ftp(connection):
     def put_file(self, local_file, remote_file):
         try:
             # Attempt to upload the file
-            self.conn.storbinary(f"STOR {remote_file}", open(local_file, "rb"))  # ruff: ignore[open-file-with-context-handler]
+            with open(local_file, "rb") as f:
+                self.conn.storbinary(f"STOR {remote_file}", f.read)
         except error_perm as error_message:
             self.logger.fail(f"Failed to upload file. Response: ({error_message})")
             return False

nxc/protocols/smb.py

@@ -2314,7 +2314,8 @@ class smb(connection):
 
         if self.args.pvk is not None:
             try:
-                self.pvkbytes = open(self.args.pvk, "rb").read()  # ruff: ignore[open-file-with-context-handler]
+                with open(self.args.pvk, "rb") as f:
+                    self.pvkbytes = f.read()
                 self.logger.success(f"Loading domain backupkey from {self.args.pvk}")
             except Exception as e:
                 self.logger.fail(str(e))