.datadiv_decode deobfuscator ghidra script

DatadivDecoder.py

# DatadivDecoder.py
# @category Data
# @menupath Tools.Deobfuscation.Decode .datadiv_decode XOR data
# @toolbar

# I wrote this script myself in Jython a few years ago and just had chatgpt port it to pyghidra

import re
import os
import traceback

from java.awt import Color
from java.awt import Toolkit
from java.awt.datatransfer import DataFlavor

from ghidra.app.decompiler import DecompInterface
from ghidra.app.plugin.core.colorizer import ColorizingService
from ghidra.program.model.address import AddressSet
from ghidra.program.model.data import StringDataType


# ---------------------------------------------------------------------
# User-tweakable defaults
# ---------------------------------------------------------------------

HIGHLIGHT_COLOR = Color(255, 255, 0)       # yellow
DECOMP_TIMEOUT_SECS = 60
FUNCTION_NAME_PREFIX = ".datadiv_decode"

# Conservative default: decode bytes and clear old data, but don't force new strings.
CREATE_STRINGS_FOR_PRINTABLE_NULL_TERMINATED_RANGES = False

# If true, process all functions whose names start with FUNCTION_NAME_PREFIX
# without prompting. Useful if you run headless.
HEADLESS_DEFAULT_PROCESS_ALL = False


# ---------------------------------------------------------------------
# Regexes
# ---------------------------------------------------------------------

SYM_RE = r"(?:uRam[0-9A-Fa-f]+|_?DAT_[0-9A-Fa-f]+)"

SINGLE_XOR_RE = re.compile(
    r"(?P<lhs>{sym})\s*=\s*(?P=lhs)\s*\^\s*(?P<imm>0x[0-9A-Fa-f]+|\d+)\s*;"
    .format(sym=SYM_RE),
    re.MULTILINE,
)

NOT_RE = re.compile(
    r"(?P<lhs>{sym})\s*=\s*~\s*(?P=lhs)\s*;"
    .format(sym=SYM_RE),
    re.MULTILINE,
)

# Handles common Ghidra decompiler output like:
#   do {
#     (&DAT_0013a000)[i] = something ^ 0xaa;
#     i = i + 1;
#   } while (i != 0x20);
#
# It is intentionally permissive about the RHS before the xor.
LOOP_XOR_RE = re.compile(
    r"do\s*\{\s*"
    r"\(&(?:DAT_)?(?P<addr>[0-9A-Fa-f]+)\)\[(?P<idx>[A-Za-z_][A-Za-z0-9_]*)\]\s*=\s*[^;]*?\^\s*(?P<imm>0x[0-9A-Fa-f]+|\d+)\s*;\s*"
    r"(?P=idx)\s*=\s*(?P=idx)\s*\+\s*1\s*;\s*"
    r"\}\s*while\s*\(\s*(?P=idx)\s*!=\s*(?P<count>0x[0-9A-Fa-f]+|\d+)\s*\)\s*;",
    re.MULTILINE | re.DOTALL,
)


# ---------------------------------------------------------------------
# Ghidra helpers
# ---------------------------------------------------------------------

program = currentProgram
listing = program.getListing()
memory = program.getMemory()
addr_factory = program.getAddressFactory()

tool = state.getTool()
colorizing_service = tool.getService(ColorizingService) if tool is not None else None


def println(msg=""):
    print(str(msg))


def parse_int(s):
    s = s.strip()
    if s.lower().startswith("0x"):
        return int(s, 16)
    return int(s, 10)


def symbol_to_address_string(sym):
    """
    Converts:
      DAT_0013a750                  -> 0013a750
      _DAT_0013a790                 -> 0013a790
      uRam000000000013a798          -> 000000000013a798
    """
    if sym.startswith("uRam"):
        return sym[len("uRam"):]
    if sym.startswith("_DAT_"):
        return sym[len("_DAT_"):]
    if sym.startswith("DAT_"):
        return sym[len("DAT_"):]
    return sym


def get_addr(addr_text):
    """
    Resolve a bare hex address against Ghidra's address factory.
    """
    addr_text = addr_text.strip()

    addr = addr_factory.getAddress(addr_text)
    if addr is not None:
        return addr

    # Fallback for default address space.
    default_space = addr_factory.getDefaultAddressSpace()
    return default_space.getAddress(int(addr_text, 16))


def imm_width_bytes(imm_text, imm_value):
    """
    Infer assignment width from immediate syntax.

    Examples:
      0xb0                       -> 1
      199                        -> 1
      0x7373737373737373         -> 8
      0x202020202020202          -> 8, because odd number of hex digits implies leading zero
    """
    s = imm_text.strip()

    if s.lower().startswith("0x"):
        hex_digits = s[2:]
        return max(1, (len(hex_digits) + 1) // 2)

    if imm_value <= 0xff:
        return 1

    return max(1, (imm_value.bit_length() + 7) // 8)


def int_to_mask_bytes(value, size):
    endian = program.getLanguage().getLanguageDescription().getEndian()
    little = not endian.isBigEndian()

    masks = []
    for i in range(size):
        shift = i * 8 if little else (size - 1 - i) * 8
        masks.append((value >> shift) & 0xff)

    return masks


def java_byte_to_int(b):
    return int(b) & 0xff


def int_to_signed_java_byte(v):
    v &= 0xff
    return v - 256 if v >= 128 else v


def get_byte(addr):
    return java_byte_to_int(memory.getByte(addr))


def put_byte(addr, value):
    """
    memory.setByte expects a signed Java byte value.
    """
    memory.setByte(addr, int_to_signed_java_byte(value))


def get_bg(addr):
    if colorizing_service is None:
        return None
    return colorizing_service.getBackgroundColor(addr)


def is_done_color(addr):
    return get_bg(addr) == HIGHLIGHT_COLOR


def set_bg(addr, color):
    if colorizing_service is None:
        return

    # Different Ghidra versions expose different overloads.
    try:
        colorizing_service.setBackgroundColor(addr, addr, color)
        return
    except Exception:
        pass

    aset = AddressSet(addr, addr)
    colorizing_service.setBackgroundColor(aset, color)


def clear_code_units_for_runs(addrs):
    """
    Clear data/code units only over uncolored target bytes.

    This avoids clobbering previously processed bytes and satisfies the
    requirement that existing datatypes on obfuscated data be cleared before
    rewriting bytes.
    """
    if not addrs:
        return

    sorted_addrs = sorted(addrs, key=lambda a: a.getOffset())

    run_start = sorted_addrs[0]
    prev = sorted_addrs[0]

    for addr in sorted_addrs[1:]:
        if addr.getOffset() == prev.getOffset() + 1:
            prev = addr
            continue

        listing.clearCodeUnits(run_start, prev, False)
        run_start = addr
        prev = addr

    listing.clearCodeUnits(run_start, prev, False)


def maybe_create_string(addr, size):
    """
    Optional helper. Disabled by default because not all decoded blobs are strings.
    Creates a StringDataType only if the bytes are printable ASCII and null-terminated.
    """
    if not CREATE_STRINGS_FOR_PRINTABLE_NULL_TERMINATED_RANGES:
        return

    try:
        vals = [get_byte(addr.add(i)) for i in range(size)]
        if not vals or vals[-1] != 0:
            return

        body = vals[:-1]
        if not body:
            return

        for b in body:
            if b not in (9, 10, 13) and (b < 0x20 or b > 0x7e):
                return

        listing.clearCodeUnits(addr, addr.add(size - 1), False)
        listing.createData(addr, StringDataType.dataType, size)
    except Exception as e:
        println("String creation skipped at {}: {}".format(addr, e))


# ---------------------------------------------------------------------
# Decoder core
# ---------------------------------------------------------------------

class DecodeOp(object):
    def __init__(self, addr, size, imm_value=None, imm_text=None, reason="", op_type="xor"):
        self.addr = addr
        self.size = size
        self.imm_value = imm_value
        self.imm_text = imm_text
        self.reason = reason
        self.op_type = op_type

    def __repr__(self):
        if self.op_type == "xor":
            op_desc = "xor={}".format(self.imm_text)
        elif self.op_type == "not":
            op_desc = "not"
        else:
            op_desc = "op={}".format(self.op_type)

        return "DecodeOp(addr={}, size={}, {}, reason={})".format(
            self.addr, self.size, op_desc, self.reason
        )
def parse_decode_ops_from_c(c_text, source_name="<text>"):
    ops = []

    for m in SINGLE_XOR_RE.finditer(c_text):
        lhs = m.group("lhs")
        imm_text = m.group("imm")
        imm_value = parse_int(imm_text)
        size = imm_width_bytes(imm_text, imm_value)

        addr_text = symbol_to_address_string(lhs)
        addr = get_addr(addr_text)

        ops.append(DecodeOp(
            addr=addr,
            size=size,
            imm_value=imm_value,
            imm_text=imm_text,
            reason="single XOR {} in {}".format(lhs, source_name),
            op_type="xor",
        ))

    for m in NOT_RE.finditer(c_text):
        lhs = m.group("lhs")

        addr_text = symbol_to_address_string(lhs)
        addr = get_addr(addr_text)

        size = 1

        ops.append(DecodeOp(
            addr=addr,
            size=size,
            reason="bitwise NOT {} in {}".format(lhs, source_name),
            op_type="not",
        ))

    for m in LOOP_XOR_RE.finditer(c_text):
        addr_text = m.group("addr")
        imm_text = m.group("imm")
        count_text = m.group("count")

        imm_value = parse_int(imm_text)
        count = parse_int(count_text)
        addr = get_addr(addr_text)

        ops.append(DecodeOp(
            addr=addr,
            size=count,
            imm_value=imm_value,
            imm_text=imm_text,
            reason="loop XOR {} bytes at {} in {}".format(count, addr_text, source_name),
            op_type="xor",
        ))

    return ops

def apply_decode_op(op):
    all_done = True
    uncolored_addrs = []

    for i in range(op.size):
        addr = op.addr.add(i)
        if not is_done_color(addr):
            all_done = False
            uncolored_addrs.append(addr)

    if all_done:
        println("SKIP already colored: {}".format(op))
        return (0, op.size, True)

    masks = None
    if op.op_type == "xor":
        masks = int_to_mask_bytes(op.imm_value, op.size)

    println("APPLY {}".format(op))

    clear_code_units_for_runs(uncolored_addrs)

    decoded_count = 0
    skipped_count = 0

    for i in range(op.size):
        addr = op.addr.add(i)

        if is_done_color(addr):
            skipped_count += 1
            continue

        old = get_byte(addr)

        if op.op_type == "xor":
            new = old ^ masks[i % len(masks)]
        elif op.op_type == "not":
            new = (~old) & 0xff
        else:
            raise RuntimeError("Unknown decode op type: {}".format(op.op_type))

        put_byte(addr, new)
        set_bg(addr, HIGHLIGHT_COLOR)
        decoded_count += 1

    maybe_create_string(op.addr, op.size)

    return (decoded_count, skipped_count, False)


# ---------------------------------------------------------------------
# Decompiler helpers
# ---------------------------------------------------------------------

def make_decompiler():
    ifc = DecompInterface()
    ifc.openProgram(program)
    return ifc


def decompile_function(ifc, func):
    result = ifc.decompileFunction(func, DECOMP_TIMEOUT_SECS, monitor)

    if result is None or not result.decompileCompleted():
        err = result.getErrorMessage() if result is not None else "no result"
        raise RuntimeError("Decompiler failed for {}: {}".format(func.getName(), err))

    return result.getDecompiledFunction().getC()


def all_datadiv_decode_functions():
    fm = program.getFunctionManager()
    funcs = []
    for f in fm.getFunctions(True):
        if f.getName().startswith(FUNCTION_NAME_PREFIX):
            funcs.append(f)
    return funcs


def function_by_name(name):
    fm = program.getFunctionManager()

    # First exact match.
    for f in fm.getFunctions(True):
        if f.getName() == name:
            return f

    # Then forgiving contains match.
    matches = []
    for f in fm.getFunctions(True):
        if name in f.getName():
            matches.append(f)

    if len(matches) == 1:
        return matches[0]

    if len(matches) > 1:
        names = "\n".join("  " + f.getName() for f in matches[:30])
        raise RuntimeError("Ambiguous function name '{}'. Matches:\n{}".format(name, names))

    raise RuntimeError("No function found matching '{}'".format(name))


def current_function():
    f = getFunctionContaining(currentAddress)
    if f is None:
        raise RuntimeError("No current function at {}".format(currentAddress))
    return f


# ---------------------------------------------------------------------
# UI/input modes
# ---------------------------------------------------------------------

def read_clipboard_text():
    clipboard = Toolkit.getDefaultToolkit().getSystemClipboard()
    data = clipboard.getData(DataFlavor.stringFlavor)
    return str(data)

from java.util import ArrayList

def java_list(items):
    out = ArrayList()
    for item in items:
        out.add(item)
    return out


def choose_mode():
    if isRunningHeadless():
        return "All .datadiv_decode functions" if HEADLESS_DEFAULT_PROCESS_ALL else "Current function"

    choices = java_list([
        "All .datadiv_decode functions",
        "Current function",
        "Function by name",
        "C text file",
        "Clipboard C text",
    ])

    default_choice = "All .datadiv_decode functions"

    return str(askChoice(
        "Decode .datadiv_decode data",
        "Input mode",
        choices,
        default_choice,
    ))


def collect_ops():
    mode = choose_mode()
    ops = []

    if mode == "All .datadiv_decode functions":
        funcs = all_datadiv_decode_functions()
        println("Found {} {}* functions".format(len(funcs), FUNCTION_NAME_PREFIX))

        ifc = make_decompiler()
        for f in funcs:
            if monitor.isCancelled():
                break

            println("Decompiling {}".format(f.getName()))
            c_text = decompile_function(ifc, f)
            ops.extend(parse_decode_ops_from_c(c_text, f.getName()))

        return ops

    if mode == "Current function":
        f = current_function()
        ifc = make_decompiler()
        c_text = decompile_function(ifc, f)
        return parse_decode_ops_from_c(c_text, f.getName())

    if mode == "Function by name":
        name = askString("Function name", "Function name or unique substring")
        f = function_by_name(name)
        ifc = make_decompiler()
        c_text = decompile_function(ifc, f)
        return parse_decode_ops_from_c(c_text, f.getName())

    if mode == "C text file":
        file_obj = askFile("Choose decompiler C text file", "Decode")
        path = file_obj.getAbsolutePath()
        with open(path, "r", encoding="utf-8", errors="replace") as fp:
            c_text = fp.read()
        return parse_decode_ops_from_c(c_text, os.path.basename(path))

    if mode == "Clipboard C text":
        c_text = read_clipboard_text()
        return parse_decode_ops_from_c(c_text, "<clipboard>")

    raise RuntimeError("Unknown mode: {}".format(mode))


# ---------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------

tx = None

try:
    if colorizing_service is None:
        raise RuntimeError("ColorizingService is unavailable. Run this inside the normal Ghidra UI tool.")

    ops = collect_ops()

    if not ops:
        println("No decode operations found.")
    else:
        println("Found {} decode operations.".format(len(ops)))

    tx = program.startTransaction("Decode .datadiv_decode XOR data")

    decoded_total = 0
    skipped_total = 0
    already_done_ops = 0

    for idx, op in enumerate(ops, 1):
        if monitor.isCancelled():
            break

        monitor.setMessage("Applying decode op {}/{} at {}".format(idx, len(ops), op.addr))

        decoded, skipped, already_done = apply_decode_op(op)
        decoded_total += decoded
        skipped_total += skipped
        if already_done:
            already_done_ops += 1

    println("")
    println("Done.")
    println("  ops found:          {}".format(len(ops)))
    println("  ops already done:   {}".format(already_done_ops))
    println("  bytes decoded:      {}".format(decoded_total))
    println("  bytes skipped:      {}".format(skipped_total))
    println("  highlight color:    rgb({}, {}, {})".format(
        HIGHLIGHT_COLOR.getRed(),
        HIGHLIGHT_COLOR.getGreen(),
        HIGHLIGHT_COLOR.getBlue(),
    ))

    program.endTransaction(tx, True)
    tx = None

except Exception as e:
    println("ERROR: {}".format(e))
    traceback.print_exc()

    if tx is not None:
        program.endTransaction(tx, False)
        tx = None

    raise