"""Dependency-free boundaries for untrusted model inputs and output.""" import json import math MAX_JSON_BYTES = 64_000 def strict_json(text): if not isinstance(text, str) or len(text.encode('utf-8')) > MAX_JSON_BYTES: raise ValueError('JSON must be text of at most 64,000 bytes') def pairs(items): result = {} for key, value in items: if key in result: raise ValueError('Duplicate JSON key') result[key] = value return result def constant(value): raise ValueError('Non-finite JSON number') try: result = json.loads(text, object_pairs_hook=pairs, parse_constant=constant) pending = [(result, 0)] while pending: value, depth = pending.pop() if depth > 32: raise ValueError('JSON nesting exceeds 32 levels') if isinstance(value, float) and not math.isfinite(value): raise ValueError('Non-finite JSON number') if isinstance(value, dict): for key in value: key.encode('utf-8') pending.extend((v, depth + 1) for v in value.values()) elif isinstance(value, list): pending.extend((v, depth + 1) for v in value) elif isinstance(value, str): value.encode('utf-8') return result except (RecursionError, UnicodeError) as exc: raise ValueError('Invalid or excessively nested JSON') from exc def validate_input(quote, schema): if not isinstance(quote, str) or not quote.strip() or len(quote) > 4000: raise ValueError('Quote must contain 1–4,000 characters') if not isinstance(schema, dict) or not 1 <= len(schema) <= 10: raise ValueError('Schema must contain 1–10 splits') for split, columns in schema.items(): if not isinstance(split, str) or not split.strip() or len(split) > 128: raise ValueError('Split names must contain 1–128 characters') if not isinstance(columns, list) or not 1 <= len(columns) <= 100: raise ValueError('Each split needs 1–100 columns') if any(not isinstance(c, str) or not c.strip() or len(c) > 128 for c in columns): raise ValueError('Column names must contain 1–128 characters') if len(columns) != len(set(columns)): raise ValueError('Duplicate schema columns') def parse_proposal(raw, quote, schema): from data_v2 import grounded from training_data import reviewed_candidate validate_input(quote, schema) target = strict_json(raw) grounded(target, schema) return reviewed_candidate(quote, quote, target)