diff --git a/litellm/proxy/guardrails/guardrail_hooks/custom_code/primitives.py b/litellm/proxy/guardrails/guardrail_hooks/custom_code/primitives.py index 2cdb65ca5d3..b2e330e218c 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/custom_code/primitives.py +++ b/litellm/proxy/guardrails/guardrail_hooks/custom_code/primitives.py @@ -224,41 +224,56 @@ def json_schema_valid(obj: Any, schema: Dict[str, Any]) -> bool: return False -def _basic_json_schema_validate(obj: Any, schema: Dict[str, Any]) -> bool: +def _basic_json_schema_validate( + obj: Any, schema: Dict[str, Any], max_depth: int = 50 +) -> bool: """ Basic JSON schema validation without external library. Handles: type, required, properties + + Uses an iterative approach with a stack to avoid recursion limits. + max_depth limits nesting to prevent infinite loops from circular schemas. """ - # Check type - schema_type = schema.get("type") - if schema_type: - type_map = { - "object": dict, - "array": list, - "string": str, - "number": (int, float), - "integer": int, - "boolean": bool, - "null": type(None), - } - expected_type = type_map.get(schema_type) - if expected_type and not isinstance(obj, expected_type): + type_map = { + "object": dict, + "array": list, + "string": str, + "number": (int, float), + "integer": int, + "boolean": bool, + "null": type(None), + } + + # Stack of (obj, schema, depth) tuples to process + stack: List[tuple] = [(obj, schema, 0)] + + while stack: + current_obj, current_schema, depth = stack.pop() + + # Circuit breaker: stop if we've gone too deep + if depth > max_depth: return False - # Check required fields - if isinstance(obj, dict): - required = schema.get("required", []) - for field in required: - if field not in obj: + # Check type + schema_type = current_schema.get("type") + if schema_type: + expected_type = type_map.get(schema_type) + if expected_type and not isinstance(current_obj, expected_type): return False - # Check properties - properties = schema.get("properties", {}) - for prop_name, prop_schema in properties.items(): - if prop_name in obj: - if not _basic_json_schema_validate(obj[prop_name], prop_schema): + # Check required fields and properties for dicts + if isinstance(current_obj, dict): + required = current_schema.get("required", []) + for field in required: + if field not in current_obj: return False + # Queue property validations + properties = current_schema.get("properties", {}) + for prop_name, prop_schema in properties.items(): + if prop_name in current_obj: + stack.append((current_obj[prop_name], prop_schema, depth + 1)) + return True diff --git a/tests/code_coverage_tests/recursive_detector.py b/tests/code_coverage_tests/recursive_detector.py index d5640f4256c..c656addd8d8 100644 --- a/tests/code_coverage_tests/recursive_detector.py +++ b/tests/code_coverage_tests/recursive_detector.py @@ -40,6 +40,7 @@ IGNORE_FUNCTIONS = [ "filter_exceptions_from_params", # max depth set (default 20) to prevent infinite recursion. "__getattr__", # lazy loading pattern in litellm/__init__.py with proper caching to prevent infinite recursion. "_validate_inheritance_chain", # max depth set (default 100) to prevent infinite recursion in policy inheritance validation. + "_basic_json_schema_validate", # max depth set. ]