Module refinery.units.malware.kramer

Classes

class kramer

Deobfuscate Python samples obfuscated with Kramer.

Expand source code Browse git

class kramer(Unit):
    """
    Deobfuscate Python samples obfuscated with Kramer.
    """
    @classmethod
    def handles(cls, data) -> bool | None:
        from refinery.lib.id import PycMagicPattern
        if PycMagicPattern.fullmatch(data[:4]):
            return True

    _LINEBREAK_MAGIC = 950

    def _crawl(self, code: CodeType, version, strings: list[str], secret: set[int], depth: int = 1):
        """
        Collects the string and integer arguments of all instructions in the given code object and
        in the code objects it contains, in the order of disassembly.
        """
        for instruction in disassemble_code(code, version):
            arg = instruction.argval
            if arg is None:
                continue
            if isinstance(arg, tuple):
                continue
            if isinstance(arg, str):
                strings.append(arg)
                continue
            if isinstance(arg, int):
                secret.add(arg)
                continue
            try:
                self._crawl(arg, version, strings, secret, depth + 1)
            except Exception as E:
                self.log_info(F'error crawling arg of type {type(arg).__name__} at depth {depth}: {E}')

    def process(self, data):
        strings: list[str] = []
        secret: set[int] = set()

        for code in extract_code_from_buffer(bytes(data)):
            self._crawl(code.container, code.version, strings, secret)

        kramer = max(strings, key=len, default='')

        if not kramer:
            raise ValueError('could not find the encoded string')

        separator = re.search('[^a-fA-F0-9]+', kramer)

        if not separator:
            raise ValueError('no separator detected; encoding method may have changed')

        def rotchar(c: int):
            if c in range(0x61, 0x7a) or c in range(0x30, 0x39):
                return c + 1
            if c == 0x7a:
                return 0x30
            if c == 0x39:
                return 0x61
            return c

        def decrypt(c: int, k: int):
            if c >= k:
                out = rotchar(c - k)
                if out not in range(0x100):
                    raise _WrongKey
                return out
            if c == self._LINEBREAK_MAGIC:
                return 0x0A
            raise _WrongKey

        def decrypt_with_key(key: int):
            decrypted = bytearray(decrypt(c, key) for c in encrypted)
            if not re.fullmatch(B'[\\s!-~]+', decrypted):
                raise _WrongKey
            return decrypted

        separator = separator.group(0)
        encrypted = [ord(bytes.fromhex(e).decode()) for e in kramer.split(separator)]

        ubound = min(x for x in encrypted if x != self._LINEBREAK_MAGIC)
        lbound = ubound - 0xFF

        secret = {k for k in secret if k > lbound and k < ubound}
        self.log_debug('potential secrets from code:', secret)

        for key in sorted(secret, reverse=True):
            try:
                return decrypt_with_key(key)
            except _WrongKey:
                pass

        self.log_info(F'all candidates failed, searching [{lbound}, {ubound}]')

        for key in range(ubound, lbound - 1, -1):
            try:
                self.log_debug('attempting key:', key)
                return decrypt_with_key(key)
            except _WrongKey:
                pass

        raise RuntimeError('could not find decryption key')

Ancestors

• Unit
• UnitBase
• Entry

Subclasses

• kramer

Class variables

var reverse

The type of the None singleton.

Inherited members

• Unit:
  • console
  • handles
  • optional_dependencies
  • required_dependencies
• Unit:
  • FilterEverything
  • Requires
  • act
  • assemble
  • carved
  • close
  • codec
  • finish
  • is_quiet
  • is_reversible
  • isatty
  • labelled
  • leniency
  • log_always
  • log_debug
  • log_detach
  • log_fail
  • log_info
  • log_level
  • log_warn
  • logger
  • name
  • nozzle
  • read
  • read1
  • reset
  • run
  • source
  • superinit
• UnitBase:
  • process
• UnitBase:
  • filter