mirror of
https://github.com/Pennyw0rth/NetExec
synced 2026-06-06 16:34:30 +00:00
ruff: manually fixed pydocstyle ruff alerts and add exclusions
This commit is contained in:
Marshall Hallenbeck
+11
-12
@@ -169,8 +169,8 @@ class connection(object):
|
||||
self.call_cmd_args()
|
||||
|
||||
def call_cmd_args(self):
|
||||
"""
|
||||
Calls all the methods specified by the command line arguments
|
||||
"""Calls all the methods specified by the command line arguments
|
||||
|
||||
Iterates over the attributes of an object (self.args)
|
||||
For each attribute, it checks if the object (self) has an attribute with the same name and if that attribute is callable (i.e., a function)
|
||||
If both conditions are met and the attribute value is not False or None,
|
||||
@@ -191,8 +191,8 @@ class connection(object):
|
||||
getattr(self, attr)()
|
||||
|
||||
def call_modules(self):
|
||||
"""
|
||||
This function calls the modules and performs various actions based on the module's attributes.
|
||||
"""Calls modules and performs various actions based on the module's attributes.
|
||||
|
||||
It iterates over the modules specified in the command line arguments.
|
||||
For each module, it loads the module and creates a context object, then calls functions based on the module's attributes.
|
||||
"""
|
||||
@@ -256,8 +256,8 @@ class connection(object):
|
||||
return False
|
||||
|
||||
def query_db_creds(self):
|
||||
"""
|
||||
Queries the database for credentials to be used for authentication.
|
||||
"""Queries the database for credentials to be used for authentication.
|
||||
|
||||
Valid cred_id values are:
|
||||
- a single cred_id
|
||||
- a range specified with a dash (ex. 1-5)
|
||||
@@ -296,8 +296,8 @@ class connection(object):
|
||||
return domains, usernames, owned, secrets, cred_types, data
|
||||
|
||||
def parse_credentials(self):
|
||||
"""
|
||||
Parse credentials from the command line or from a file specified.
|
||||
r"""Parse credentials from the command line or from a file specified.
|
||||
|
||||
Usernames can be specified with a domain (domain\\username) or without (username).
|
||||
If the file contains domain\\username the domain specified will be overwritten by the one in the file.
|
||||
|
||||
@@ -381,8 +381,8 @@ class connection(object):
|
||||
return domain, username, owned, secret, cred_type, [None] * len(secret)
|
||||
|
||||
def try_credentials(self, domain, username, owned, secret, cred_type, data=None):
|
||||
"""
|
||||
Try to login using the specified credentials and protocol.
|
||||
"""Try to login using the specified credentials and protocol.
|
||||
|
||||
Possible login methods are:
|
||||
- plaintext (/kerberos)
|
||||
- NTLM-hash (/kerberos)
|
||||
@@ -415,8 +415,7 @@ class connection(object):
|
||||
return self.kerberos_login(domain, username, "", "", secret, self.kdcHost, False)
|
||||
|
||||
def login(self):
|
||||
"""
|
||||
Try to login using the credentials specified in the command line or in the database.
|
||||
"""Try to login using the credentials specified in the command line or in the database.
|
||||
|
||||
:return: True if the login was successful and "--continue-on-success" was not specified, False otherwise.
|
||||
"""
|
||||
|
||||
@@ -6,8 +6,7 @@ from neo4j.exceptions import AuthError, ServiceUnavailable
|
||||
|
||||
|
||||
def add_user_bh(user, domain, logger, config):
|
||||
"""
|
||||
Adds a user to the BloodHound graph database.
|
||||
"""Adds a user to the BloodHound graph database.
|
||||
|
||||
Args:
|
||||
----
|
||||
|
||||
+4
-6
@@ -45,12 +45,10 @@ def called_from_cmd_args():
|
||||
|
||||
# Stolen from https://github.com/pydanny/whichcraft/
|
||||
def which(cmd, mode=os.F_OK | os.X_OK, path=None):
|
||||
"""Given a command, mode, and a PATH string, return the path which
|
||||
conforms to the given mode on the PATH, or None if there is no such
|
||||
file.
|
||||
`mode` defaults to os.F_OK | os.X_OK. `path` defaults to the result
|
||||
of os.environ.get("PATH"), or can be overridden with a custom search
|
||||
path.
|
||||
"""Find the path which conforms to the given mode on the PATH for a command.
|
||||
|
||||
Given a command, mode, and a PATH string, return the path which conforms to the given mode on the PATH, or None if there is no such file.
|
||||
`mode` defaults to os.F_OK | os.X_OK. `path` defaults to the result of os.environ.get("PATH"), or can be overridden with a custom search path.
|
||||
Note: This function was backported from the Python 3 source code.
|
||||
"""
|
||||
|
||||
|
||||
@@ -15,8 +15,7 @@ obfuscate_ps_scripts = False
|
||||
|
||||
|
||||
def get_ps_script(path):
|
||||
"""
|
||||
Generates a full path to a PowerShell script given a relative path.
|
||||
"""Generates a full path to a PowerShell script given a relative path.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
|
||||
+5
-4
@@ -37,8 +37,8 @@ class NXCAdapter(logging.LoggerAdapter):
|
||||
# logging.getLogger("impacket").disabled = True
|
||||
|
||||
def format(self, msg, *args, **kwargs):
|
||||
"""
|
||||
Format msg for output if needed
|
||||
"""Format msg for output
|
||||
|
||||
This is used instead of process() since process() applies to _all_ messages, including debug calls
|
||||
"""
|
||||
if self.extra is None:
|
||||
@@ -125,9 +125,10 @@ class NXCAdapter(logging.LoggerAdapter):
|
||||
self.log_console_to_file(text, *args, **kwargs)
|
||||
|
||||
def log_console_to_file(self, text, *args, **kwargs):
|
||||
"""
|
||||
"""Log the console output to a file
|
||||
|
||||
If debug or info logging is not enabled, we still want display/success/fail logged to the file specified,
|
||||
so we create a custom LogRecord and pass it to all the additional handlers (which will be all the file handlers
|
||||
so we create a custom LogRecord and pass it to all the additional handlers (which will be all the file handlers)
|
||||
"""
|
||||
if self.logger.getEffectiveLevel() >= logging.INFO:
|
||||
# will be 0 if it's just the console output, so only do this if we actually have file loggers
|
||||
|
||||
@@ -21,7 +21,7 @@ class NXCModule:
|
||||
multiple_hosts = True
|
||||
|
||||
def options(self, context, module_options):
|
||||
"""
|
||||
r"""
|
||||
TMP_DIR Path where process dump should be saved on target system (default: C:\\Windows\\Temp\\)
|
||||
HANDLEKATZ_PATH Path where handlekatz.exe is on your system (default: /tmp/)
|
||||
HANDLEKATZ_EXE_NAME Name of the handlekatz executable (default: handlekatz.exe)
|
||||
|
||||
@@ -20,7 +20,7 @@ class NXCModule:
|
||||
self.search_path = "'C:\\Users\\','$env:PROGRAMFILES','env:ProgramFiles(x86)'"
|
||||
|
||||
def options(self, context, module_options):
|
||||
"""
|
||||
r"""
|
||||
SEARCH_TYPE Specify what to search, between:
|
||||
PROCESS Look for running KeePass.exe process only
|
||||
FILES Look for KeePass-related files (KeePass.config.xml, .kdbx, KeePass.exe) only, may take some time
|
||||
|
||||
@@ -56,7 +56,7 @@ class NXCModule:
|
||||
self.restart_keepass_script_str = restart_keepass_script_file.read()
|
||||
|
||||
def options(self, context, module_options):
|
||||
"""
|
||||
r"""
|
||||
ACTION (mandatory) Performs one of the following actions, specified by the user:
|
||||
ADD insert a new malicious trigger into KEEPASS_CONFIG_PATH's specified file
|
||||
CHECK check if a malicious trigger is currently set in KEEPASS_CONFIG_PATH's
|
||||
|
||||
@@ -13,7 +13,7 @@ class NXCModule:
|
||||
multiple_hosts = True
|
||||
|
||||
def options(self, context, module_options):
|
||||
"""
|
||||
r"""
|
||||
CA Certificate Authority Name (CA_SERVER\CA_NAME)
|
||||
TEMPLATE Template name allowing users to authenticate with (default: User)
|
||||
DC_IP IP Address of the domain controller
|
||||
|
||||
@@ -34,7 +34,7 @@ class NXCModule:
|
||||
self.module_options = module_options
|
||||
|
||||
def options(self, context, module_options):
|
||||
"""
|
||||
r"""
|
||||
TMP_DIR Path where process dump should be saved on target system (default: C:\\Windows\\Temp\\)
|
||||
NANO_PATH Path where nano.exe is on your system (default: OS temp directory)
|
||||
NANO_EXE_NAME Name of the nano executable (default: nano.exe)
|
||||
|
||||
+1
-1
@@ -11,7 +11,7 @@ class NXCModule:
|
||||
multiple_hosts = True
|
||||
|
||||
def options(self, context, module_options):
|
||||
"""
|
||||
r"""
|
||||
PID // Process ID for Target User, PID=pid
|
||||
EXEC // Command to exec, EXEC='command' Single quote is better to use
|
||||
|
||||
|
||||
@@ -20,7 +20,7 @@ class NXCModule:
|
||||
multiple_hosts = True
|
||||
|
||||
def options(self, context, module_options):
|
||||
"""
|
||||
r"""
|
||||
TMP_DIR Path where process dump should be saved on target system (default: C:\\Windows\\Temp\\)
|
||||
PROCDUMP_PATH Path where procdump.exe is on your system (default: /tmp/), if changed embeded version will not be used
|
||||
PROCDUMP_EXE_NAME Name of the procdump executable (default: procdump.exe), if changed embeded version will not be used
|
||||
|
||||
+27
-41
@@ -15,10 +15,7 @@ CHUNK_SIZE = 4096
|
||||
|
||||
|
||||
def human_size(nbytes):
|
||||
"""
|
||||
This function takes a number of bytes as input and converts it to a human-readable
|
||||
size representation with appropriate units (e.g., KB, MB, GB, TB).
|
||||
"""
|
||||
"""Takes a number of bytes as input and converts it to a human-readable size representation with appropriate units (e.g., KB, MB, GB, TB)"""
|
||||
suffixes = ["B", "KB", "MB", "GB", "TB", "PB", "EB", "ZB", "YB"]
|
||||
|
||||
# Find the appropriate unit suffix and convert bytes to higher units
|
||||
@@ -35,17 +32,12 @@ def human_size(nbytes):
|
||||
|
||||
|
||||
def human_time(timestamp):
|
||||
"""This function takes a numerical timestamp (seconds since the epoch) and formats it
|
||||
as a human-readable date and time in the format "YYYY-MM-DD HH:MM:SS".
|
||||
"""
|
||||
"""Takes a numerical timestamp (seconds since the epoch) and formats it as a human-readable date and time in the format "YYYY-MM-DD HH:MM:SS"""
|
||||
return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(timestamp))
|
||||
|
||||
|
||||
def make_dirs(path):
|
||||
"""
|
||||
This function attempts to create directories at the given path. It handles the
|
||||
exception `os.errno.EEXIST` that may occur if the directories already exist.
|
||||
"""
|
||||
"""Creates directories at the given path. It handles the exception `os.errno.EEXIST` that may occur if the directories already exist."""
|
||||
try:
|
||||
os.makedirs(path)
|
||||
except OSError as e:
|
||||
@@ -55,8 +47,7 @@ def make_dirs(path):
|
||||
|
||||
|
||||
def get_list_from_option(opt):
|
||||
"""
|
||||
This function takes a comma-separated string and converts it to a list of lowercase strings.
|
||||
"""Takes a comma-separated string and converts it to a list of lowercase strings.
|
||||
It filters out empty strings from the input before converting.
|
||||
"""
|
||||
return list(map(lambda o: o.lower(), filter(bool, opt.split(","))))
|
||||
@@ -106,9 +97,8 @@ class SMBSpiderPlus:
|
||||
make_dirs(self.output_folder)
|
||||
|
||||
def reconnect(self):
|
||||
"""This function performs a series of reconnection attempts, up to `self.max_connection_attempts`,
|
||||
with a 3-second delay between each attempt. It renegotiates the session by creating a new
|
||||
connection object and logging in again.
|
||||
"""Performs a series of reconnection attempts, up to `self.max_connection_attempts`, with a 3-second delay between each attempt.
|
||||
It renegotiates the session by creating a new connection object and logging in again.
|
||||
"""
|
||||
for i in range(1, self.max_connection_attempts + 1):
|
||||
self.logger.display(f"Reconnection attempt #{i}/{self.max_connection_attempts} to server.")
|
||||
@@ -122,7 +112,7 @@ class SMBSpiderPlus:
|
||||
return False
|
||||
|
||||
def list_path(self, share, subfolder):
|
||||
"""This function returns a list of paths for a given share/folder."""
|
||||
"""Returns a list of paths for a given share/folder."""
|
||||
filelist = []
|
||||
try:
|
||||
# Get file list for the current folder
|
||||
@@ -144,7 +134,7 @@ class SMBSpiderPlus:
|
||||
return filelist
|
||||
|
||||
def get_remote_file(self, share, path):
|
||||
"""This function will check if a path is readable in a SMB share."""
|
||||
"""Checks if a path is readable in a SMB share."""
|
||||
try:
|
||||
remote_file = RemoteFile(self.smb.conn, path, share, access=FILE_READ_DATA)
|
||||
return remote_file
|
||||
@@ -155,10 +145,9 @@ class SMBSpiderPlus:
|
||||
return None
|
||||
|
||||
def read_chunk(self, remote_file, chunk_size=CHUNK_SIZE):
|
||||
"""This function reads the next chunk of data from the provided remote file using
|
||||
the specified chunk size. If a `SessionError` is encountered,
|
||||
it retries up to 3 times by reconnecting the SMB connection. If the maximum number
|
||||
of retries is exhausted or an unexpected exception occurs, it returns an empty chunk.
|
||||
"""Reads the next chunk of data from the provided remote file using the specified chunk size.
|
||||
If a `SessionError` is encountered, it retries up to 3 times by reconnecting the SMB connection.
|
||||
If the maximum number of retries is exhausted or an unexpected exception occurs, it returns an empty chunk.
|
||||
"""
|
||||
chunk = ""
|
||||
retry = 3
|
||||
@@ -182,9 +171,9 @@ class SMBSpiderPlus:
|
||||
return chunk
|
||||
|
||||
def get_file_save_path(self, remote_file):
|
||||
"""This function processes the remote file path to extract the filename and the folder
|
||||
path where the file should be saved locally. It converts forward slashes (/) and backslashes (\)
|
||||
in the remote file path to the appropriate path separator for the local file system.
|
||||
r"""Processes the remote file path to extract the filename and the folder path where the file should be saved locally.
|
||||
|
||||
It converts forward slashes (/) and backslashes (\) in the remote file path to the appropriate path separator for the local file system.
|
||||
The folder path and filename are then obtained separately.
|
||||
"""
|
||||
# Remove the backslash before the remote host part and replace slashes with the appropriate path separator
|
||||
@@ -199,9 +188,7 @@ class SMBSpiderPlus:
|
||||
return folder, filename
|
||||
|
||||
def spider_shares(self):
|
||||
"""This function enumerates all available shares for the SMB connection, spiders
|
||||
through the readable shares, and saves the metadata of the shares to a JSON file.
|
||||
"""
|
||||
"""Enumerates all available shares for the SMB connection, spiders through the readable shares, and saves the metadata of the shares to a JSON file"""
|
||||
self.logger.info("Enumerating shares for spidering.")
|
||||
shares = self.smb.shares()
|
||||
|
||||
@@ -251,9 +238,9 @@ class SMBSpiderPlus:
|
||||
return self.results
|
||||
|
||||
def spider_folder(self, share_name, folder):
|
||||
"""This recursive function traverses through the contents of the specified share and folder.
|
||||
It checks each entry (file or folder) against various filters, performs file metadata recording,
|
||||
and downloads eligible files if the download flag is set.
|
||||
"""Traverses through the contents of the specified share and folder.
|
||||
|
||||
It checks each entry (file or folder) against various filters, performs file metadata recording, and downloads eligible files if the download flag is set.
|
||||
"""
|
||||
self.logger.info(f'Spider share "{share_name}" in folder "{folder}".')
|
||||
|
||||
@@ -284,9 +271,7 @@ class SMBSpiderPlus:
|
||||
self.parse_file(share_name, next_fullpath, result)
|
||||
|
||||
def parse_file(self, share_name, file_path, file_info):
|
||||
"""This function checks file attributes against various filters, records file metadata,
|
||||
and downloads eligible files if the download flag is set.
|
||||
"""
|
||||
"""Checks file attributes against various filters, records file metadata, and downloads eligible files if the download flag is set"""
|
||||
# Record the file metadata
|
||||
file_size = file_info.get_filesize()
|
||||
file_creation_time = file_info.get_ctime_epoch()
|
||||
@@ -361,7 +346,8 @@ class SMBSpiderPlus:
|
||||
self.stats["num_get_fail"] += 1
|
||||
|
||||
def save_file(self, remote_file, share_name):
|
||||
"""This function reads the `remote_file` in chunks using the `read_chunk` method.
|
||||
"""Reads the `remote_file` in chunks using the `read_chunk` method.
|
||||
|
||||
Each chunk is then written to the local file until the entire file is saved.
|
||||
It handles cases where the file remains empty due to errors.
|
||||
"""
|
||||
@@ -392,9 +378,9 @@ class SMBSpiderPlus:
|
||||
self.logger.fail(f'Unable to download file "{remote_path}".')
|
||||
|
||||
def dump_folder_metadata(self, results):
|
||||
"""This function takes the metadata results as input and writes them to a JSON file
|
||||
in the `self.output_folder`. The results are formatted with indentation and
|
||||
sorted keys before being written to the file.
|
||||
"""Takes the metadata results as input and writes them to a JSON file in the `self.output_folder`.
|
||||
|
||||
The results are formatted with indentation and sorted keys before being written to the file.
|
||||
"""
|
||||
metadata_path = os.path.join(self.output_folder, f"{self.host}.json")
|
||||
try:
|
||||
@@ -405,7 +391,7 @@ class SMBSpiderPlus:
|
||||
self.logger.fail(f"Failed to save share metadata: {str(e)}")
|
||||
|
||||
def print_stats(self):
|
||||
"""This function prints the statistics during processing."""
|
||||
"""Prints the statistics during processing"""
|
||||
# Share statistics.
|
||||
shares = self.stats.get("shares", [])
|
||||
if shares:
|
||||
@@ -493,8 +479,8 @@ class SMBSpiderPlus:
|
||||
|
||||
|
||||
class NXCModule:
|
||||
"""
|
||||
Spider plus module
|
||||
"""Spider Plus Nodule
|
||||
|
||||
Module by @vincd
|
||||
Updated by @godylockz
|
||||
"""
|
||||
|
||||
@@ -27,7 +27,7 @@ class NXCModule:
|
||||
multiple_hosts = True
|
||||
|
||||
def options(self, context, module_options):
|
||||
"""
|
||||
r"""
|
||||
PATH Specify the Path if you already found a WinSCP.ini file. (Example: PATH="C:\\Users\\USERNAME\\Documents\\WinSCP_Passwords\\WinSCP.ini")
|
||||
|
||||
REQUIRES ADMIN PRIVILEGES:
|
||||
|
||||
+1
-1
@@ -45,7 +45,7 @@ class RequestHandler(BaseHTTPRequestHandler):
|
||||
self.server.module.on_response(self.server.context, self)
|
||||
|
||||
def stop_tracking_host(self):
|
||||
"""This gets called when a module has finshed executing, removes the host from the connection tracker list"""
|
||||
"""Called when a module has finshed executing, removes the host from the connection tracker list"""
|
||||
try:
|
||||
self.server.hosts.remove(self.client_address[0])
|
||||
if hasattr(self.server.module, "on_shutdown"):
|
||||
|
||||
+1
-1
@@ -81,7 +81,7 @@ build-backend = "poetry.core.masonry.api"
|
||||
# McCabe complexity (`C901`) by default.
|
||||
# Other options: N (pep8-naming), D (pydocstyle)
|
||||
select = ["E", "F", "D"]
|
||||
ignore = [ "E501", "F405", "F841", "D100", "D101", "D102", "D103", "D104", "D105", "D106", "D107", "D203", "D204", "D212", "D213", "D400", "D415", "D419"]
|
||||
ignore = [ "E501", "F405", "F841", "D100", "D101", "D102", "D103", "D104", "D105", "D106", "D107", "D203", "D204", "D205", "D212", "D213", "D400", "D401", "D415", "D417", "D419"]
|
||||
|
||||
# Allow autofix for all enabled rules (when `--fix`) is provided.
|
||||
fixable = ["ALL"]
|
||||
|
||||
Reference in New Issue
Block a user