Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 13 additions & 12 deletions src/rat_king_parser/config_parser/rat_config_parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -81,21 +81,22 @@ def __init__(
if data is None and not isfile(file_path):
raise ConfigParserException("File not found")
# Filled in _decrypt_and_decode_config()
self._incompatible_decryptors: list[Any] = []
self._dnpp = DotNetPEPayload(file_path, yara_rule, data)
self._incompatible_decryptors: list[int] = []
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The type hint for _incompatible_decryptors is list[int], but the code appends decryptor classes to this list (e.g., on line 144: self._incompatible_decryptors.append(decryptor)), not integers. This is a type mismatch. The previous type hint list[Any] was more appropriate.

Suggested change
self._incompatible_decryptors: list[int] = []
self._incompatible_decryptors: list[Any] = []

try:
self._dnpp = DotNetPEPayload(file_path, yara_rule, data)
except Exception as e:
raise e
Comment on lines +85 to +88
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

This try...except block is redundant. Catching an exception only to immediately re-raise it provides no additional functionality and adds unnecessary clutter to the code. It can be safely removed.

            self._dnpp = DotNetPEPayload(file_path, yara_rule, data)

self.report["sha256"] = self._dnpp.sha256
self.report["yara_possible_family"] = self._dnpp.yara_match

# Assigned in _decrypt_and_decode_config()
self._decryptor: ConfigDecryptor = None
self.report["config"] = self._get_config()
key_hex = "None"
if self._decryptor is not None and self._decryptor.key is not None:
if isinstance(self._decryptor.key, bytes):
key_hex = self._decryptor.key.hex()
else:
key_hex = str(self._decryptor.key)
self.report["key"] = key_hex
self.report["key"] = (
self._decryptor.key.hex()
if self._decryptor is not None and self._decryptor.key is not None
else "None"
)
Comment on lines +95 to +99
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

This logic for setting self.report["key"] is incorrect. It unconditionally calls .hex() on self._decryptor.key. However, self._decryptor.key can be a str (as defined in ConfigDecryptor), which does not have a .hex() method. This will lead to an AttributeError at runtime. The previous implementation correctly handled both bytes and other types and was more robust.

            key_hex = "None"
            if self._decryptor is not None and self._decryptor.key is not None:
                if isinstance(self._decryptor.key, bytes):
                    key_hex = self._decryptor.key.hex()
                else:
                    key_hex = str(self._decryptor.key)
            self.report["key"] = key_hex

self.report["salt"] = (
self._decryptor.salt.hex()
if self._decryptor is not None and self._decryptor.salt is not None
Expand All @@ -121,7 +122,7 @@ def _decrypt_and_decode_config(
config_fields_map[k] = field_name
item_data[field_name] = v
if len(item_data) > 0:
if isinstance(item, config_item.EncryptedStringConfigItem):
if type(item) is config_item.EncryptedStringConfigItem:
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Using type(item) is ... is generally discouraged in Python in favor of isinstance(item, ...). isinstance correctly handles inheritance, so if a subclass of EncryptedStringConfigItem were ever used, this check would still work. Using type makes the code less robust and extensible.

                if isinstance(item, config_item.EncryptedStringConfigItem):

# Translate config value RVAs to string values
for k in item_data:
item_data[k] = self._dnpp.user_string_from_rva(item_data[k])
Expand Down Expand Up @@ -158,7 +159,7 @@ def _decrypt_and_decode_config(
if self._decryptor is None:
raise ConfigParserException("All decryptors failed")

elif isinstance(item, config_item.ByteArrayConfigItem):
elif type(item) is config_item.ByteArrayConfigItem:
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Similar to a previous comment, using type(item) is ... is less robust than isinstance(item, ...). isinstance should be preferred as it correctly handles subclasses, making the code more maintainable and extensible.

                elif isinstance(item, config_item.ByteArrayConfigItem):

for k in item_data:
arr_size, arr_rva = item_data[k]
item_data[k] = self._dnpp.byte_array_from_size_and_rva(
Expand All @@ -167,7 +168,7 @@ def _decrypt_and_decode_config(

decoded_config.update(item_data)
# UrlHost is a marker of a special case until this can be standardized
if len(decoded_config) < min_config_len and "UrlHost" not in decoded_config:
if len(decoded_config) < min_config_len and "UrlHost" not in item_data:
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

This check is likely incorrect. It checks if "UrlHost" is in item_data, but item_data only contains the data for the last config item processed in the loop. The check should be against decoded_config, which accumulates all parsed config items throughout the loop.

        if len(decoded_config) < min_config_len and "UrlHost" not in decoded_config:

raise ConfigParserException(
f"Minimum threshold of config items not met: {len(decoded_config)}/{min_config_len}"
)
Expand Down
12 changes: 6 additions & 6 deletions src/rat_king_parser/config_parser/utils/data_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,11 +31,11 @@


# Converts a bytes object to an int object using the specified byte order
def bytes_to_int(data: bytes, order: str = "little") -> int:
def bytes_to_int(bytes: bytes, order: str = "little") -> int:
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The parameter bytes shadows the Python built-in bytes type. This is considered bad practice as it can lead to confusion and potential bugs. A more descriptive name like data or byte_string should be used.

Suggested change
def bytes_to_int(bytes: bytes, order: str = "little") -> int:
def bytes_to_int(data: bytes, order: str = "little") -> int:

try:
return int.from_bytes(data, byteorder=order)
return int.from_bytes(bytes, byteorder=order)
except Exception:
raise ConfigParserException(f"Error parsing int from value: {data}")
raise ConfigParserException(f"Error parsing int from value: {bytes}")


# Decodes a bytes object to a Unicode string, using UTF-16LE for byte values
Expand All @@ -57,8 +57,8 @@ def decode_bytes(byte_str: bytes | str) -> str:


# Converts an int to a bytes object, with the specified length and order
def int_to_bytes(value: int, length: int = 4, order: str = "little") -> bytes:
def int_to_bytes(int: int, length: int = 4, order: str = "little") -> bytes:
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The parameter int shadows the Python built-in int type. This is bad practice and can make the code harder to read and maintain. A different name like value or integer should be used.

Suggested change
def int_to_bytes(int: int, length: int = 4, order: str = "little") -> bytes:
def int_to_bytes(value: int, length: int = 4, order: str = "little") -> bytes:

try:
return value.to_bytes(length, order)
return int.to_bytes(length, order)
except Exception:
raise ConfigParserException(f"Error parsing bytes from value: {value}")
raise ConfigParserException(f"Error parsing bytes from value: {int}")
Original file line number Diff line number Diff line change
Expand Up @@ -174,7 +174,6 @@ def decrypt_encrypted_strings(

# Extracts AES key candidates from the payload
def _get_aes_key_candidates(self, encrypted_strings: dict[str, str]) -> list[bytes]:

logger.debug("Extracting AES key candidates...")
keys = []

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,7 @@ def _decrypt(self, iv: bytes, ciphertext: bytes) -> bytes:
)

cipher = AES.new(self.key, mode=CBC, iv=iv)
unpadded_text = ""

Comment on lines +70 to 71
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The initialization unpadded_text = "" is redundant. The variable is always assigned a new value within the try block, and if that fails, an exception is raised, so this initial value is never used. This line and the following blank line can be removed for clarity.

try:
unpadded_text = cipher.decrypt(ciphertext)
Expand Down Expand Up @@ -112,7 +113,7 @@ def decrypt_encrypted_strings(
result = decode_bytes(self._decrypt(self.iv, decoded_val))
except ConfigParserException as e:
last_exc = e
logger.debug(f"Error decrypting {k}: {e}")
print("error", e)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Using print() for error reporting is not a good practice in a library or application. It bypasses the configured logging system, making it difficult to control output (e.g., verbosity, format, destination). It's better to use the logging module, as was done in the previous version of the code.

Suggested change
print("error", e)
logger.debug(f"Error decrypting {k}: {e}")


if result is None:
logger.debug(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -79,9 +79,9 @@ def _get_hardcoded_hosts(self) -> list[str]:
hardcoded_hosts = []
for rva in hardcoded_host_rvas:
try:
hardcoded_host = self._payload.user_string_from_rva(bytes_to_int(rva))
if hardcoded_host != ".":
hardcoded_hosts.append(hardcoded_host)
harcoded_host = self._payload.user_string_from_rva(bytes_to_int(rva))
if harcoded_host != ".":
hardcoded_hosts.append(harcoded_host)
Comment on lines +82 to +84
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

There is a typo in the variable name harcoded_host. It should be hardcoded_host. This typo is repeated on the following lines.

Suggested change
harcoded_host = self._payload.user_string_from_rva(bytes_to_int(rva))
if harcoded_host != ".":
hardcoded_hosts.append(harcoded_host)
hardcoded_host = self._payload.user_string_from_rva(bytes_to_int(rva))
if hardcoded_host != ".":
hardcoded_hosts.append(hardcoded_host)

except Exception as e:
logger.error(f"Error translating hardcoded host at {hex(rva)}: {e}")
continue
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
from logging import getLogger
import logging
from base64 import b64decode
from hashlib import md5
from re import DOTALL, compile, search
Expand All @@ -42,7 +42,7 @@
from ..dotnetpe_payload import DotNetPEPayload
from .config_decryptor import ConfigDecryptor, IncompatibleDecryptorException

logger = getLogger(__name__)
logger = logging.getLogger(__name__)


# Is old AES - specifically Rijndael in CBC mode with MD5 hashing for key derivation
Expand Down Expand Up @@ -110,14 +110,11 @@ def decrypt_encrypted_strings(self, encrypted_strings: dict[str, str]) -> dict[s
else:
for key_pattern in self._KEY_PATTERNS:
key_hit = search(key_pattern, self._payload.data)
if not key_hit:
continue
key_rva = bytes_to_int(key_hit.groups()[0])
raw_key_field = self._payload.field_name_from_rva(key_rva)
if raw_key_field in encrypted_strings:
key = encrypted_strings[raw_key_field]
self.key = self._derive_key(key)
break
key = encrypted_strings[raw_key_field]
self.key = self._derive_key(key)
break
Comment on lines 112 to +117
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

critical

This change introduces a critical bug by removing necessary safety checks. Without if not key_hit: continue, the code will raise an AttributeError if search() returns None. Similarly, removing if raw_key_field in encrypted_strings: will cause a KeyError if the key is not found. The previous implementation handled these cases correctly and should be restored to prevent runtime crashes.

Suggested change
key_hit = search(key_pattern, self._payload.data)
if not key_hit:
continue
key_rva = bytes_to_int(key_hit.groups()[0])
raw_key_field = self._payload.field_name_from_rva(key_rva)
if raw_key_field in encrypted_strings:
key = encrypted_strings[raw_key_field]
self.key = self._derive_key(key)
break
key = encrypted_strings[raw_key_field]
self.key = self._derive_key(key)
break
key_hit = search(key_pattern, self._payload.data)
if not key_hit:
continue
key_rva = bytes_to_int(key_hit.groups()[0])
raw_key_field = self._payload.field_name_from_rva(key_rva)
if raw_key_field in encrypted_strings:
key = encrypted_strings[raw_key_field]
self.key = self._derive_key(key)
break

except Exception as e:
raise ConfigParserException(f"Failed to derive AES key: {e}")

Expand Down
31 changes: 11 additions & 20 deletions src/rat_king_parser/config_parser/utils/dotnetpe_payload.py
Original file line number Diff line number Diff line change
Expand Up @@ -234,13 +234,6 @@ def custom_attribute_from_type(self, typespacename: str, typename: str) -> dict:
"""
config = {}
try:
ca_map = {}
for ca in self.dotnetpe.net.mdtables.CustomAttribute.rows:
idx = ca.Parent.row_index
if idx not in ca_map:
ca_map[idx] = []
ca_map[idx].append(ca)

for td in self.dotnetpe.net.mdtables.TypeDef.rows:
if (
td.TypeNamespace.value != typespacename
Expand All @@ -251,21 +244,19 @@ def custom_attribute_from_type(self, typespacename: str, typename: str) -> dict:
self.dotnetpe.net.mdtables.Property.rows
):
if pd.Name.value.startswith(
(
"Boolean_",
"BorderStyle_",
"Color_",
"Byte",
"Int32_",
"SizeF_",
"String_",
)
"Boolean_",
"BorderStyle_",
"Color_",
"Byte",
"Int32_",
"SizeF_",
"String_",
):
Comment on lines 246 to 254
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

critical

The startswith() method is being called with multiple string arguments. This is incorrect syntax and will cause a TypeError at runtime. The startswith() method accepts a single string or a tuple of strings as its first argument.

Suggested change
if pd.Name.value.startswith(
(
"Boolean_",
"BorderStyle_",
"Color_",
"Byte",
"Int32_",
"SizeF_",
"String_",
)
"Boolean_",
"BorderStyle_",
"Color_",
"Byte",
"Int32_",
"SizeF_",
"String_",
):
if pd.Name.value.startswith(
(
"Boolean_",
"BorderStyle_",
"Color_",
"Byte",
"Int32_",
"SizeF_",
"String_",
)
):

continue
# CustomAttribute Parent index is 1-based
target_index = pd_row_index + 1
if target_index in ca_map:
for ca in ca_map[target_index]:
for ca in self.dotnetpe.net.mdtables.CustomAttribute.rows:
if (
ca.Parent.row_index == pd_row_index + 1
): # CustomAttribute Parent index is 1-based
Comment on lines +256 to +259
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

This nested loop introduces a significant performance regression. By removing the pre-computed ca_map and instead iterating over all CustomAttribute.rows for each Property, the complexity of this section increases from roughly O(N+M) to O(N*M). For files with many properties and custom attributes, this will be noticeably slower. The previous implementation using a map was much more efficient and should be restored.

# Extract the value from the CustomAttribute blob
blob_data = ca.Value.value
if blob_data and blob_data != b"\x01\x00\x00\x00":
Expand Down