style: Automatic code formatting

Author: actions-user

lib/parsers_aux/ratking/__init__.py

@@ -34,17 +34,14 @@
 from re import DOTALL, compile, search
 from typing import Any, Tuple
 
-# from yara import Rules
-
 from .config_parser_exception import ConfigParserException
 from .utils import config_item
-from .utils.decryptors import (
-    SUPPORTED_DECRYPTORS,
-    ConfigDecryptor,
-    IncompatibleDecryptorException,
-)
+from .utils.decryptors import SUPPORTED_DECRYPTORS, ConfigDecryptor, IncompatibleDecryptorException
 from .utils.dotnetpe_payload import DotNetPEPayload
 
+# from yara import Rules
+
+
 logger = getLogger(__name__)
 
 
@@ -54,9 +51,7 @@ class RATConfigParser:
     _MIN_CONFIG_LEN_CEILING = 9
 
     # Pattern to find the VerifyHash() method
-    _PATTERN_VERIFY_HASH = compile(
-        rb"\x7e.{3}\x04(?:\x6f.{3}\x0a){2}\x74.{3}\x01", DOTALL
-    )
+    _PATTERN_VERIFY_HASH = compile(rb"\x7e.{3}\x04(?:\x6f.{3}\x0a){2}\x74.{3}\x01", DOTALL)
 
     # def __init__(self, file_path: str, yara_rule: Rules = None) -> None:
     def __init__(self, file_data: bytes = None) -> None:
@@ -77,33 +72,24 @@ def __init__(self, file_data: bytes = None) -> None:
             self._decryptor: ConfigDecryptor = None
             self.report["config"] = self._get_config()
             self.report["key"] = (
-                self._decryptor.key.hex()
-                if self._decryptor is not None and self._decryptor.key is not None
-                else "None"
+                self._decryptor.key.hex() if self._decryptor is not None and self._decryptor.key is not None else "None"
             )
             self.report["salt"] = (
-                self._decryptor.salt.hex()
-                if self._decryptor is not None and self._decryptor.salt is not None
-                else "None"
+                self._decryptor.salt.hex() if self._decryptor is not None and self._decryptor.salt is not None else "None"
             )
         except Exception as e:
             # self.report["config"] = f"Exception encountered for {file_path}: {e}"
             self.report["config"] = f"Exception encountered: {e}"
 
     # Decrypts/decodes values from an encrypted config and returns the
     # decrypted/decoded config
-    def _decrypt_and_decode_config(
-        self, encrypted_config: bytes, min_config_len: int
-    ) -> dict[str, Any]:
+    def _decrypt_and_decode_config(self, encrypted_config: bytes, min_config_len: int) -> dict[str, Any]:
         decoded_config = {}
 
         for item_class in config_item.SUPPORTED_CONFIG_ITEMS:
             item = item_class()
             # Translate config Field RVAs to Field names
-            item_data = {
-                self._dnpp.field_name_from_rva(k): v
-                for k, v in item.parse_from(encrypted_config).items()
-            }
+            item_data = {self._dnpp.field_name_from_rva(k): v for k, v in item.parse_from(encrypted_config).items()}
 
             if len(item_data) > 0:
                 if type(item) is config_item.EncryptedStringConfigItem:
@@ -122,22 +108,16 @@ def _decrypt_and_decode_config(
                             try:
                                 self._decryptor = decryptor(self._dnpp)
                             except IncompatibleDecryptorException as ide:
-                                logger.debug(
-                                    f"Decryptor incompatible {decryptor} : {ide}"
-                                )
+                                logger.debug(f"Decryptor incompatible {decryptor} : {ide}")
                                 self._incompatible_decryptors.append(decryptor)
                                 continue
                         try:
                             # Try to decrypt the encrypted strings
                             # Continue to next compatible decryptor on failure
-                            item_data = self._decryptor.decrypt_encrypted_strings(
-                                item_data
-                            )
+                            item_data = self._decryptor.decrypt_encrypted_strings(item_data)
                             break
                         except Exception as e:
-                            logger.debug(
-                                f"Decryption failed with decryptor {decryptor} : {e}"
-                            )
+                            logger.debug(f"Decryption failed with decryptor {decryptor} : {e}")
                             self._decryptor = None
 
                     if self._decryptor is None:
@@ -146,16 +126,12 @@ def _decrypt_and_decode_config(
                 elif type(item) is 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(
-                            arr_size, arr_rva
-                        ).hex()
+                        item_data[k] = self._dnpp.byte_array_from_size_and_rva(arr_size, arr_rva).hex()
 
                 decoded_config.update(item_data)
 
         if len(decoded_config) < min_config_len:
-            raise ConfigParserException(
-                f"Minimum threshold of config items not met: {len(decoded_config)}/{min_config_len}"
-            )
+            raise ConfigParserException(f"Minimum threshold of config items not met: {len(decoded_config)}/{min_config_len}")
         return decoded_config
 
     # Searches for the RAT configuration section, using the VerifyHash() marker
@@ -185,38 +161,29 @@ def _get_config_cctor_brute_force(self) -> Tuple[int, dict[str, Any]]:
             raise ConfigParserException("No .cctor method could be found")
 
         # For each .cctor method, map its RVA and body (in raw bytes)
-        candidate_cctor_data = {
-            method.rva: self._dnpp.method_body_from_method(method)
-            for method in candidates
-        }
+        candidate_cctor_data = {method.rva: self._dnpp.method_body_from_method(method) for method in candidates}
 
         config_start, decrypted_config = None, None
         # Start at our ceiling value for number of config items
         min_config_len = self._MIN_CONFIG_LEN_CEILING
 
         while decrypted_config is None and min_config_len >= self._MIN_CONFIG_LEN_FLOOR:
             for method_rva, method_body in candidate_cctor_data.items():
-                logger.debug(
-                    f"Attempting brute force at .cctor method at {hex(method_rva)}"
-                )
+                logger.debug(f"Attempting brute force at .cctor method at {hex(method_rva)}")
                 try:
                     config_start, decrypted_config = (
                         method_rva,
                         self._decrypt_and_decode_config(method_body, min_config_len),
                     )
                     break
                 except Exception as e:
-                    logger.debug(
-                        f"Brute force failed for method at {hex(method_rva)}: {e}"
-                    )
+                    logger.debug(f"Brute force failed for method at {hex(method_rva)}: {e}")
                     continue
             # Reduce the minimum config length until we reach our floor
             min_config_len -= 1
 
         if decrypted_config is None:
-            raise ConfigParserException(
-                "No valid configuration could be parsed from any .cctor methods"
-            )
+            raise ConfigParserException("No valid configuration could be parsed from any .cctor methods")
         return config_start, decrypted_config
 
     # Attempts to retrieve the config via looking for a config section preceded
@@ -230,16 +197,12 @@ def _get_config_verify_hash_method(self) -> Tuple[int, dict[str, Any]]:
 
         # Reverse the hit to find the VerifyHash() method, then grab the
         # subsequent function
-        config_method = self._dnpp.method_from_instruction_offset(
-            verify_hash_hit.start(), 1
-        )
+        config_method = self._dnpp.method_from_instruction_offset(verify_hash_hit.start(), 1)
         encrypted_config = self._dnpp.method_body_from_method(config_method)
         min_config_len = self._MIN_CONFIG_LEN_CEILING
         while True:
             try:
-                decrypted_config = self._decrypt_and_decode_config(
-                    encrypted_config, min_config_len
-                )
+                decrypted_config = self._decrypt_and_decode_config(encrypted_config, min_config_len)
                 return config_method.rva, decrypted_config
             except Exception as e:
                 # Reduce the minimum config length until we reach our floor

lib/parsers_aux/ratking/utils/config_item.py

@@ -68,9 +68,7 @@ def parse_from(self, data: bytes) -> dict[int, Any]:
                 fields[field_rva] = field_value
                 found_items += 1
             else:
-                logger.debug(
-                    f"Overlapping Field RVAs detected in config at {hex(field_rva)}"
-                )
+                logger.debug(f"Overlapping Field RVAs detected in config at {hex(field_rva)}")
         logger.debug(f"Parsed {found_items} {self._label} values")
         return fields
 

lib/parsers_aux/ratking/utils/data_utils.py

@@ -50,9 +50,7 @@ def decode_bytes(byte_str: bytes | str) -> str:
         else:
             result = byte_str.decode("utf-8")
     except Exception:
-        raise ConfigParserException(
-            f"Error decoding bytes object to Unicode: {byte_str}"
-        )
+        raise ConfigParserException(f"Error decoding bytes object to Unicode: {byte_str}")
     return result
 
 

lib/parsers_aux/ratking/utils/decryptors/config_decryptor.py

@@ -49,7 +49,5 @@ def __init__(self, payload: DotNetPEPayload) -> None:
     # Abstract method to take in a map representing a configuration of config
     # Field names and values and return a decoded/decrypted configuration
     @abstractmethod
-    def decrypt_encrypted_strings(
-        self, encrypted_strings: dict[str, str]
-    ) -> dict[str, list[str] | str]:
+    def decrypt_encrypted_strings(self, encrypted_strings: dict[str, str]) -> dict[str, list[str] | str]:
         pass

lib/parsers_aux/ratking/utils/decryptors/config_decryptor_aes_cbc.py

@@ -56,9 +56,7 @@ class ConfigDecryptorAESCBC(ConfigDecryptor):
     _MIN_CIPHERTEXT_LEN = 48
 
     # Patterns for identifying AES metadata
-    _PATTERN_AES_KEY_AND_BLOCK_SIZE = compile(
-        b"[\x06-\x09]\x20(.{4})\x6f.{4}[\x06-\x09]\x20(.{4})", DOTALL
-    )
+    _PATTERN_AES_KEY_AND_BLOCK_SIZE = compile(b"[\x06-\x09]\x20(.{4})\x6f.{4}[\x06-\x09]\x20(.{4})", DOTALL)
     # Do not compile in-line replacement patterns
     _PATTERN_AES_KEY_BASE = b"(.{3}\x04).%b"
     _PATTERN_AES_SALT_INIT = b"\x80%b\x2a"
@@ -79,9 +77,7 @@ def __init__(self, payload: DotNetPEPayload) -> None:
     # Given an initialization vector and ciphertext, creates a Cipher
     # object with the AES key and specified IV and decrypts the ciphertext
     def _decrypt(self, iv: bytes, ciphertext: bytes) -> bytes:
-        logger.debug(
-            f"Decrypting {ciphertext} with key {self.key.hex()} and IV {iv.hex()}..."
-        )
+        logger.debug(f"Decrypting {ciphertext} with key {self.key.hex()} and IV {iv.hex()}...")
         aes_cipher = Cipher(AES(self.key), CBC(iv), backend=default_backend())
         decryptor = aes_cipher.decryptor()
         # Use a PKCS7 unpadder to remove padding from decrypted value
@@ -113,9 +109,7 @@ def _derive_aes_passphrase_candidates(self, key_val: str) -> list[bytes]:
         return passphrase_candidates
 
     # Decrypts encrypted config values with the provided cipher data
-    def decrypt_encrypted_strings(
-        self, encrypted_strings: dict[str, str]
-    ) -> dict[str, str]:
+    def decrypt_encrypted_strings(self, encrypted_strings: dict[str, str]) -> dict[str, str]:
         logger.debug("Decrypting encrypted strings...")
         if self._key_candidates is None:
             self._key_candidates = self._get_aes_key_candidates(encrypted_strings)
@@ -157,9 +151,7 @@ def decrypt_encrypted_strings(
                     last_exc = e
 
             if result is None:
-                logger.debug(
-                    f"Decryption failed for item {v}: {last_exc}; Leaving as original value..."
-                )
+                logger.debug(f"Decryption failed for item {v}: {last_exc}; Leaving as original value...")
                 result = v
 
             logger.debug(f"Key: {k}, Value: {result}")
@@ -174,9 +166,7 @@ def _get_aes_key_candidates(self, encrypted_strings: dict[str, str]) -> list[byt
         keys = []
 
         # Use the key Field name to index into our existing config
-        key_raw_value = encrypted_strings[
-            self._payload.field_name_from_rva(self._key_rva)
-        ]
+        key_raw_value = encrypted_strings[self._payload.field_name_from_rva(self._key_rva)]
         passphrase_candidates = self._derive_aes_passphrase_candidates(key_raw_value)
 
         for candidate in passphrase_candidates:
@@ -196,9 +186,7 @@ def _get_aes_key_candidates(self, encrypted_strings: dict[str, str]) -> list[byt
                 continue
 
         if len(keys) == 0:
-            raise ConfigParserException(
-                f"Could not derive key from passphrase candidates: {passphrase_candidates}"
-            )
+            raise ConfigParserException(f"Could not derive key from passphrase candidates: {passphrase_candidates}")
         return keys
 
     # Extracts the AES key and block size from the payload
@@ -222,9 +210,7 @@ def _get_aes_key_rva(self, metadata_ins_offset: int) -> int:
         logger.debug("Extracting AES key RVA...")
 
         # Get the RVA of the method that sets up AES256 metadata
-        metadata_method_token = self._payload.method_from_instruction_offset(
-            metadata_ins_offset, by_token=True
-        ).token
+        metadata_method_token = self._payload.method_from_instruction_offset(metadata_ins_offset, by_token=True).token
 
         # Insert this RVA into the KEY_BASE pattern to find where the AES key
         # is initialized
@@ -269,9 +255,7 @@ def _get_aes_salt(self, salt_rva: int) -> bytes:
         #
         # stsfld	uint8[] Client.Algorithm.Aes256::Salt
         # ret
-        aes_salt_initialization = self._payload.data.find(
-            self._PATTERN_AES_SALT_INIT % escape(salt_rva)
-        )
+        aes_salt_initialization = self._payload.data.find(self._PATTERN_AES_SALT_INIT % escape(salt_rva))
         if aes_salt_initialization == -1:
             raise ConfigParserException("Could not identify AES salt initialization")
 
@@ -283,9 +267,7 @@ def _get_aes_salt(self, salt_rva: int) -> bytes:
         salt_op = bytes([self._payload.data[salt_op_offset]])
 
         # Get the salt RVA from the 4 bytes following the initialization op
-        salt_strings_rva_packed = self._payload.data[
-            salt_op_offset + 1 : salt_op_offset + 5
-        ]
+        salt_strings_rva_packed = self._payload.data[salt_op_offset + 1 : salt_op_offset + 5]
         salt_strings_rva = bytes_to_int(salt_strings_rva_packed)
 
         # If the op is a ldstr op, just get the bytes value of the string being
@@ -300,12 +282,9 @@ def _get_aes_salt(self, salt_rva: int) -> bytes:
         # byte array value from the FieldRVA table
         elif salt_op == OPCODE_LDTOKEN:
             salt_size = self._payload.data[salt_op_offset - 7]
-            salt = self._payload.byte_array_from_size_and_rva(
-                salt_size, salt_strings_rva
-            )
+            salt = self._payload.byte_array_from_size_and_rva(salt_size, salt_strings_rva)
         else:
             raise ConfigParserException(f"Unknown salt opcode found: {salt_op.hex()}")
 
         logger.debug(f"Found salt value: {salt.hex()}")
         return salt
-

lib/parsers_aux/ratking/utils/decryptors/config_decryptor_aes_ecb.py

@@ -74,17 +74,13 @@ def _decrypt(self, ciphertext: bytes) -> bytes:
             padded_text = decryptor.update(ciphertext) + decryptor.finalize()
             unpadded_text = unpadder.update(padded_text) + unpadder.finalize()
         except Exception as e:
-            raise ConfigParserException(
-                f"Error decrypting ciphertext {ciphertext} with key {self.key.hex()}: {e}"
-            )
+            raise ConfigParserException(f"Error decrypting ciphertext {ciphertext} with key {self.key.hex()}: {e}")
 
         logger.debug(f"Decryption result: {unpadded_text}")
         return unpadded_text
 
     # Decrypts encrypted config values with the provided cipher data
-    def decrypt_encrypted_strings(
-        self, encrypted_strings: dict[str, str]
-    ) -> dict[str, str]:
+    def decrypt_encrypted_strings(self, encrypted_strings: dict[str, str]) -> dict[str, str]:
         logger.debug("Decrypting encrypted strings...")
 
         if self.key is None:

lib/parsers_aux/ratking/utils/decryptors/config_decryptor_decrypt_xor.py

@@ -83,9 +83,7 @@ def _decode_encoded_strings(self) -> list[str]:
 
     # Parses the config, adds decoded XOR strings, and returns the decoded
     # config
-    def decrypt_encrypted_strings(
-        self, encrypted_strings: dict[str, str]
-    ) -> dict[str, list[str] | str]:
+    def decrypt_encrypted_strings(self, encrypted_strings: dict[str, str]) -> dict[str, list[str] | str]:
         config = {}
         # Pass off plaintext config to a ConfigDecryptorPlaintext
         ptcd = ConfigDecryptorPlaintext(self._payload)
@@ -105,18 +103,13 @@ def _get_xor_metadata(self):
         self._xor_strings = list(
             filter(
                 None,
-                [
-                    self._payload.user_string_from_rva(bytes_to_int(rva))
-                    for rva in xor_string_rvas
-                ],
+                [self._payload.user_string_from_rva(bytes_to_int(rva)) for rva in xor_string_rvas],
             )
         )
         logger.debug(f"{len(self._xor_strings)} XOR strings found")
 
         # Get the static constructor containing the XOR key
-        xor_key_cctor = self._payload.method_from_instruction_offset(
-            dxor_block.start(), step=1, by_token=True
-        )
+        xor_key_cctor = self._payload.method_from_instruction_offset(dxor_block.start(), step=1, by_token=True)
         xor_key_cctor_body = self._payload.method_body_from_method(xor_key_cctor)
 
         # Derive the XOR key RVA and value

lib/parsers_aux/ratking/utils/decryptors/config_decryptor_plaintext.py

@@ -116,13 +116,9 @@ def __init__(self, payload: DotNetPEPayload) -> None:
 
     # Calculates whether the config meets the minimum threshold for known Field
     # Names and returns it if it does
-    def decrypt_encrypted_strings(
-        self, encrypted_strings: dict[str, str]
-    ) -> dict[str, str]:
+    def decrypt_encrypted_strings(self, encrypted_strings: dict[str, str]) -> dict[str, str]:
         field_names = set(encrypted_strings.keys())
         num_overlapping_field_names = len(KNOWN_CONFIG_FIELD_NAMES & field_names)
         if num_overlapping_field_names < self.MIN_THRESHOLD_MATCH:
-            raise ConfigParserException(
-                "Plaintext threshold of known config items not met"
-            )
+            raise ConfigParserException("Plaintext threshold of known config items not met")
         return encrypted_strings