Java Malware Deobfuscator for Ratty variant

checksum_decryptor.py

#!/usr/bin/env python3
"""
checksum_decryptor.py
=====================
Decrypts the encrypted config file ("checksum") embedded in this RAT's JAR.

HOW THE ENCRYPTION WORKS
─────────────────────────
The RAT stores its C2 configuration (host, port, autostart, etc.) as an
encrypted file named literally "checksum" inside the JAR's root classpath.

Step-by-step flow inside Config.loadConfig():

  1. LOCATE:   getClass().getClassLoader().getResourceAsStream("checksum")
               The file is loaded as a raw InputStream from inside the JAR.

  2. READ:     A helper converts the stream to a String — the file content is
               a Base64-encoded string (all ASCII, no line breaks).

  3. DECRYPT:  decryptAES(content, "checksum") is called.
               This is CryptUtil.decryptAES(), which works as follows:

                 a. KEY DERIVATION  — CryptUtil.generate16BitKey(keyString):
                      key = SHA-1(keyString.getBytes("UTF-8"))  →  20 bytes
                      key = Arrays.copyOf(key, 16)              →  16 bytes
                    So the AES key is always the first 16 bytes of the
                    SHA-1 digest of the key string.
                    For "checksum": SHA-1("checksum")[:16]
                      = 4a 2a 18 09 53 29 9a cf 5f 3e 24 9d 9e 93 d6 ee

                 b. CIPHER:  AES / ECB / PKCS5Padding
                      javax.crypto.Cipher.getInstance("AES/ECB/PKCS5Padding")
                      cipher.init(DECRYPT_MODE, SecretKeySpec(key, "AES"))
                      plaintext = cipher.doFinal(Base64.decode(content))

  4. PARSE:    The plaintext is a standard Java .properties file, loaded with
               Properties.load(new ByteArrayInputStream(plaintext)).

  5. FIELDS:   The property keys are themselves obfuscated strings decrypted
               at class-init time. Resolved keys are:
                 Host               → C2 hostname or IP
                 Port               → C2 TCP port (integer)
                 AutoStart          → install persistence on boot (boolean)
                 Hide_Client_File   → mark client file as hidden (boolean)
                 Show_Message_Box   → show lure dialog on first run (boolean)
                 Message_Box_Title  → lure dialog title
                 Message_Box_Text   → lure dialog body text
                 Message_Box_Category → JOptionPane message type (-1/0/1/2/3)

WHY "checksum"?
───────────────
Using the filename "checksum" is a social-engineering trick — security scanners
and analysts often skip files with that name assuming they're integrity hashes.
The AES key is also "checksum", making the whole scheme self-referential and
easy for the attacker to remember without storing the key anywhere obvious.

Usage:
    python3 checksum_decryptor.py --jar malware.jar
    python3 checksum_decryptor.py --jar malware.jar --checksum-file ./checksum
    python3 checksum_decryptor.py --checksum-file ./checksum

Requirements:
    pip install pycryptodome
"""

import argparse
import base64
import hashlib
import io
import sys
import zipfile
from typing import Optional

from Crypto.Cipher import AES


# ─── Key derivation (mirrors CryptUtil.generate16BitKey) ─────────────────────

def derive_aes_key(key_string: str) -> bytes:
    """
    Replicates CryptUtil.generate16BitKey(keyString):
        SHA-1(keyString.getBytes("UTF-8"))  →  take first 16 bytes
    Returns a 16-byte AES-128 key.
    """
    sha1 = hashlib.sha1(key_string.encode("utf-8")).digest()  # 20 bytes
    return sha1[:16]                                           # AES-128


# ─── Decryption (mirrors CryptUtil.decryptAES) ───────────────────────────────

def decrypt_config(encoded_content: bytes, key_string: str = "checksum") -> str:
    """
    Replicates CryptUtil.decryptAES(content, keyString):
        1. Base64-decode the content
        2. AES/ECB/PKCS5Padding decrypt with derive_aes_key(keyString)
        3. Return as UTF-8 string
    """

    key        = derive_aes_key(key_string)
    ciphertext = base64.b64decode(encoded_content.strip())
    cipher     = AES.new(key, AES.MODE_ECB)
    padded     = cipher.decrypt(ciphertext)

    # Remove PKCS5 padding
    pad = padded[-1]
    plaintext = padded[:-pad] if 1 <= pad <= 16 else padded

    return plaintext.decode("utf-8", errors="replace")


# ─── Config parser (mirrors Properties.load) ─────────────────────────────────

def parse_properties(text: str) -> dict:
    """
    Parse a Java .properties file into a dict.
    Handles # comments, blank lines, and key=value pairs.
    """
    props = {}
    for line in text.splitlines():
        line = line.strip()
        if not line or line.startswith("#") or line.startswith("!"):
            continue
        if "=" in line:
            k, _, v = line.partition("=")
            props[k.strip()] = v.strip()
    return props


# ─── Locate checksum inside JAR ───────────────────────────────────────────────

def extract_checksum_from_jar(jar_path: str) -> Optional[bytes]:
    """
    Reads the 'checksum' resource from the root of the JAR classpath,
    mirroring getClassLoader().getResourceAsStream("checksum").
    """
    with zipfile.ZipFile(jar_path) as zf:
        for candidate in ("checksum", "checksum/", "META-INF/checksum"):
            if candidate in zf.namelist():
                return zf.read(candidate)
    return None


# ─── Pretty printer ───────────────────────────────────────────────────────────

def print_config(props: dict, key_string: str) -> None:
    print()
    print("╔══════════════════════════════════════════╗")
    print("║         DECRYPTED C2 CONFIGURATION       ║")
    print("╚══════════════════════════════════════════╝")
    print(f"  AES key string : {key_string!r}")
    print(f"  AES key (hex)  : {derive_aes_key(key_string).hex()}")
    print()

    # Highlight the most important fields first
    priority = ["Host", "Port", "AutoStart", "Hide_Client_File",
                "Show_Message_Box", "Message_Box_Title",
                "Message_Box_Text", "Message_Box_Category"]

    shown = set()
    for k in priority:
        if k in props:
            print(f"  {k:<25s} = {props[k]}")
            shown.add(k)

    # Any extra fields not in the known list
    for k, v in props.items():
        if k not in shown:
            print(f"  {k:<25s} = {v}")

    print()

# ─── CLI ──────────────────────────────────────────────────────────────────────

def main() -> None:
    parser = argparse.ArgumentParser(
        description="Decrypt the 'checksum' C2 config file from a RAT JAR.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__,
    )
    parser.add_argument(
        "--jar",
        metavar="FILE",
        help="JAR file to extract 'checksum' from (optional if --checksum-file given)",
    )
    parser.add_argument(
        "--checksum-file",
        metavar="FILE",
        help="Path to the raw 'checksum' file extracted from the JAR",
    )
    parser.add_argument(
        "--key",
        metavar="STRING",
        default="checksum",
        help="AES key string (default: 'checksum')",
    )
    parser.add_argument(
        "--raw",
        action="store_true",
        help="Print raw decrypted plaintext instead of parsed properties",
    )
    args = parser.parse_args()

    if not args.jar and not args.checksum_file:
        parser.error("Provide at least one of --jar or --checksum-file")

    # 1. Get the encoded content
    encoded: Optional[bytes] = None

    if args.checksum_file:
        try:
            encoded = open(args.checksum_file, "rb").read()
            print(f"[+] Loaded checksum file: {args.checksum_file}")
        except FileNotFoundError:
            print(f"[-] Checksum file not found: {args.checksum_file}", file=sys.stderr)
            sys.exit(1)

    elif args.jar:
        try:
            encoded = extract_checksum_from_jar(args.jar)
        except (FileNotFoundError, zipfile.BadZipFile) as e:
            print(f"[-] JAR error: {e}", file=sys.stderr)
            sys.exit(1)

        if encoded is None:
            print(f"[-] 'checksum' resource not found inside {args.jar}", file=sys.stderr)
            print("    Try extracting it manually and use --checksum-file", file=sys.stderr)
            sys.exit(1)

        print(f"[+] Extracted 'checksum' from JAR: {args.jar}")

    # 2. Decrypt
    try:
        plaintext = decrypt_config(encoded, key_string=args.key)
    except Exception as e:
        print(f"[-] Decryption failed: {e}", file=sys.stderr)
        sys.exit(1)

    print(f"[+] Decrypted successfully ({len(encoded)} bytes → {len(plaintext)} chars)")

    # 3. Output
    if args.raw:
        print("\n── Raw plaintext ──────────────────────────")
        print(plaintext)
        return

    props = parse_properties(plaintext)
    if not props:
        print("[-] No properties found in decrypted output. Try --raw to inspect.")
        print(plaintext)
        return

    print_config(props, key_string=args.key)


if __name__ == "__main__":
    main()

jar_string_decryptor.py

#!/usr/bin/env python3
"""
jar_string_decryptor.py
=======================
Automatically extracts and decrypts obfuscated strings from Java JAR files
that use the ZKM-style (Zelix KlassMaster) encryption pattern:

  - Blowfish/ECB  — key = MD5(key_string), full 16-byte digest
  - DES/ECB       — key = MD5(key_string)[:8]
  - XOR           — Base64-decode then XOR with repeating key string

Usage:
    python3 jar_string_decryptor.py <file.jar> [--show-garbled] [--class-filter <pattern>]

Examples:
    python3 jar_string_decryptor.py malware.jar
    python3 jar_string_decryptor.py malware.jar --class-filter com/proj/client
    python3 jar_string_decryptor.py malware.jar --show-garbled

Requirements:
    pip install pycryptodome
"""

import argparse
import base64
import hashlib
import re
import struct
import sys
import zipfile
from dataclasses import dataclass
from typing import Optional

from Crypto.Cipher import DES, Blowfish


# ─── Crypto helpers ───────────────────────────────────────────────────────────

def _unpad(data: bytes) -> bytes:
    """Remove PKCS5/PKCS7 padding."""
    if not data:
        return data
    pad = data[-1]
    if 1 <= pad <= 16:
        return data[:-pad]
    return data.rstrip(b"\x00")


def decrypt_blowfish(ct_b64: str, key_str: str) -> str:
    """Blowfish/ECB — key = MD5(key_string), full 16-byte digest.
    Only attempted when ciphertext is block-aligned (multiple of 8 bytes)."""
    key = hashlib.md5(key_str.encode("utf-8")).digest()
    ct  = base64.b64decode(ct_b64)
    if len(ct) % 8 != 0:
        raise ValueError("Blowfish requires block-aligned ciphertext")
    pt  = _unpad(Blowfish.new(key, Blowfish.MODE_ECB).decrypt(ct))
    return pt.decode("utf-8", errors="replace")


def decrypt_des(ct_b64: str, key_str: str) -> str:
    """DES/ECB — key = MD5(key_string)[:8].
    Only attempted when ciphertext is block-aligned (multiple of 8 bytes)."""
    key = hashlib.md5(key_str.encode("utf-8")).digest()[:8]
    ct  = base64.b64decode(ct_b64)
    if len(ct) % 8 != 0:
        raise ValueError("DES requires block-aligned ciphertext")
    pt  = _unpad(DES.new(key, DES.MODE_ECB).decrypt(ct))
    return pt.decode("utf-8", errors="replace")


def decrypt_xor(ct_b64: str, key_str: str) -> str:
    """XOR — Base64-decode, then XOR bytes against repeating key_string.
    Works on any ciphertext length — no block alignment needed."""
    ct = base64.b64decode(ct_b64)
    return "".join(chr(b ^ ord(key_str[i % len(key_str)])) for i, b in enumerate(ct))


# Cipher cascade: block ciphers first (guarded by alignment check inside each
# function), XOR last as it accepts any length.
CIPHERS = [
    ("blowfish", decrypt_blowfish),
    ("des",      decrypt_des),
    ("xor",      decrypt_xor),
]


# ─── Class file constant-pool parser ──────────────────────────────────────────

def parse_utf8_constants(data: bytes) -> list:
    """Return all UTF-8 constant-pool entries from a .class file, in order."""
    if data[:4] != b"\xca\xfe\xba\xbe":
        return []
    pos = 8
    count = struct.unpack(">H", data[pos : pos + 2])[0]
    pos += 2
    strings = []
    i = 1
    while i < count:
        tag = data[pos]; pos += 1
        if tag == 1:                          # Utf8
            length = struct.unpack(">H", data[pos : pos + 2])[0]; pos += 2
            strings.append(data[pos : pos + length].decode("utf-8", errors="replace"))
            pos += length
        elif tag in (5, 6):                   # Long / Double — two slots
            strings.append(None); strings.append(None)
            pos += 8; i += 1
        elif tag in (3, 4):                   # Integer / Float
            strings.append(None); pos += 4
        elif tag in (9, 10, 11, 12, 17, 18):  # Fieldref / Methodref / etc.
            strings.append(None); pos += 4
        elif tag in (7, 8, 16, 19, 20):       # Class / String / etc.
            strings.append(None); pos += 2
        elif tag == 15:                       # MethodHandle
            strings.append(None); pos += 3
        else:
            strings.append(None)
        i += 1
    return [s for s in strings if s is not None]


# ─── Heuristics ───────────────────────────────────────────────────────────────

# Min 4 alphanum chars before optional '=' padding.
# {4,} not {8,}: short ciphertexts like 'FBE0' (-> 'AES'), 'XiskIg==' (-> '.enc'),
# 'Hz4iSmM=' (-> 'SHA-1') were silently dropped with the stricter minimum.
_B64_RE = re.compile(r"^[A-Za-z0-9+/]{4,}={0,2}$")

_SKIP_STRINGS = frozenset({
    "Blowfish", "DES", "MD5", "SHA-1", "AES", "UTF_8", "UTF-8",
    "Code", "init", "read", "write", "clone", "start", "length",
    "get", "put", "run", "main", "close", "flush",
})


def looks_like_ciphertext(s: str) -> bool:
    """
    True if s could be a Base64-encoded ciphertext.

    Block-alignment is NOT required here — short XOR-encrypted strings like
    'SHA-1' (5 bytes decoded) or '.enc' (4 bytes decoded) do not align to
    8-byte DES/Blowfish blocks. Those ciphers guard their own alignment
    internally and raise ValueError, causing the cascade to fall through to XOR.
    """
    if not _B64_RE.match(s):
        return False
    try:
        return len(base64.b64decode(s)) >= 2
    except Exception:
        return False


def looks_like_key(s: str) -> bool:
    """True if s looks like a short ZKM obfuscation key (4-8 alphanum chars)."""
    return (
        bool(re.match(r"^[A-Za-z0-9]{4,8}$", s))
        and s not in _SKIP_STRINGS
    )


def is_printable(s: str) -> bool:
    """True if s is plausibly readable plaintext (>=90% printable ASCII)."""
    if not s:
        return False
    printable = sum(0x20 <= ord(c) < 0x7F or c in "\t\n\r" for c in s)
    return printable / len(s) >= 0.90


# ─── Result dataclass ─────────────────────────────────────────────────────────

@dataclass
class DecryptedString:
    class_name: str
    cipher:     str
    key:        str
    plaintext:  str
    garbled:    bool


# ─── Core scanner ─────────────────────────────────────────────────────────────

def scan_jar(jar_path: str, class_filter: str = None) -> list:
    """
    Scan every .class file in jar_path, find (ciphertext, key) consecutive
    pairs in the constant pool, and attempt decryption with all three ciphers.
    """
    results = []

    with zipfile.ZipFile(jar_path) as zf:
        class_files = [
            name for name in zf.namelist()
            if name.endswith(".class")
            and (class_filter is None or class_filter in name)
        ]

        for name in sorted(class_files):
            data     = zf.read(name)
            strings  = parse_utf8_constants(data)
            cls_name = name.replace(".class", "")

            for i in range(len(strings) - 1):
                ct_candidate  = strings[i]
                key_candidate = strings[i + 1]

                if not looks_like_ciphertext(ct_candidate):
                    continue
                if not looks_like_key(key_candidate):
                    continue

                best_cipher    = None
                best_plaintext = None

                for cipher_name, fn in CIPHERS:
                    try:
                        pt = fn(ct_candidate, key_candidate)
                        if is_printable(pt):
                            best_cipher    = cipher_name
                            best_plaintext = pt
                            break
                    except Exception:
                        continue

                if best_plaintext is not None:
                    results.append(DecryptedString(
                        class_name = cls_name,
                        cipher     = best_cipher,
                        key        = key_candidate,
                        plaintext  = best_plaintext,
                        garbled    = False,
                    ))
                else:
                    try:
                        garbled_pt = decrypt_xor(ct_candidate, key_candidate)
                    except Exception:
                        garbled_pt = "<e>"
                    results.append(DecryptedString(
                        class_name = cls_name,
                        cipher     = "?",
                        key        = key_candidate,
                        plaintext  = garbled_pt,
                        garbled    = True,
                    ))

    return results


# ─── Output formatter ─────────────────────────────────────────────────────────

def print_results(results: list, show_garbled: bool) -> None:
    visible = [r for r in results if not r.garbled or show_garbled]
    if not visible:
        print("No decrypted strings found.")
        return

    current_class = None
    for r in visible:
        if r.class_name != current_class:
            current_class = r.class_name
            print()
            print("=" * 70)
            print(f"  {current_class}")
            print("=" * 70)
        tag = "  [GARBLED]" if r.garbled else ""
        print(f"  ({r.cipher:8s}) key={r.key!r:8}  =>  {r.plaintext!r}{tag}")

    clean   = sum(1 for r in results if not r.garbled)
    garbled = sum(1 for r in results if r.garbled)
    print()
    print(f"Total: {clean} decrypted  |  {garbled} garbled (use --show-garbled to display)")


# ─── CLI ──────────────────────────────────────────────────────────────────────

def main() -> None:
    parser = argparse.ArgumentParser(
        description="Decrypt ZKM-obfuscated strings from a Java JAR file.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__,
    )
    parser.add_argument("jar", help="Path to the JAR file to scan")
    parser.add_argument(
        "--show-garbled",
        action="store_true",
        help="Also print strings that could not be cleanly decrypted",
    )
    parser.add_argument(
        "--class-filter",
        metavar="PATTERN",
        default=None,
        help="Only scan classes whose path contains PATTERN (e.g. com/proj/client)",
    )
    args = parser.parse_args()

    print(f"Scanning {args.jar} ...")
    if args.class_filter:
        print(f"Class filter: {args.class_filter!r}")

    try:
        results = scan_jar(args.jar, class_filter=args.class_filter)
    except FileNotFoundError:
        print(f"Error: file not found: {args.jar}", file=sys.stderr)
        sys.exit(1)
    except zipfile.BadZipFile:
        print(f"Error: not a valid ZIP/JAR file: {args.jar}", file=sys.stderr)
        sys.exit(1)

    print_results(results, show_garbled=args.show_garbled)


if __name__ == "__main__":
    main()