diff --git a/test/test_management_report.py b/test/test_management_report.py index 8c7803ea8..e50850703 100644 --- a/test/test_management_report.py +++ b/test/test_management_report.py @@ -244,6 +244,47 @@ class ManagementTests(unittest.TestCase): self.assertNotIn("Traceback", errors.getvalue()) password_prompt.assert_not_called() + def test_cli_rejects_deep_mqtt_jsonl_before_password_prompt(self): + path = Path(self.work.name) / "cli-deep-jsonl.json" + deep = '{"metadata":' * 10000 + "0" + "}" * 10000 + for text in ("{}\n" + deep, deep + "\n{}"): + path.write_text(text, encoding="ascii") + output, errors = io.StringIO(), io.StringIO() + with mock.patch.object(report.getpass, "getpass") as password_prompt, \ + mock.patch.object(sys, "argv", ["report.py", str(path), "--mqtt"]), \ + contextlib.redirect_stdout(output), contextlib.redirect_stderr(errors): + with self.assertRaises(SystemExit) as failure: + report.main() + self.assertEqual(failure.exception.code, 2) + self.assertEqual(output.getvalue(), "") + self.assertIn("nested too deeply", errors.getvalue()) + self.assertNotIn("Traceback", errors.getvalue()) + password_prompt.assert_not_called() + + def test_input_json_depth_limit_is_explicit_and_checks_before_decoding(self): + limit = report.MAX_JSON_DEPTH + for opening, closing in (("[", "]"), ('{"metadata":', "}")): + with self.subTest(opening=opening): + value = report._load_input_json(opening * limit + "0" + closing * limit) + for _ in range(limit): + value = value[0] if isinstance(value, list) else value["metadata"] + self.assertEqual(value, 0) + text = opening * (limit + 1) + "0" + closing * (limit + 1) + with mock.patch.object(report.json, "loads") as decoder: + with self.assertRaisesRegex(ValueError, "nested too deeply"): + report._load_input_json(text) + decoder.assert_not_called() + + def test_input_json_depth_ignores_string_brackets_and_escaped_quotes(self): + text = 'brackets [{]} and quote " and backslash \\ ' * 1000 + document = {"metadata": text, "raw": "\\"} + self.assertEqual(report._load_input_json(json.dumps(document)), document) + # An escaped backslash before a closing quote must not hide the next + # actual container, including when the following record exceeds depth. + deep = "[" * report.MAX_JSON_DEPTH + "0" + "]" * report.MAX_JSON_DEPTH + with self.assertRaisesRegex(ValueError, "nested too deeply"): + report._load_input_json("[" + json.dumps("\\") + "," + deep + "]") + def test_interoperable_full_report(self): decoded = report.decode_report(list(reversed(self.pages)), "management test password") self.assertEqual(decoded["sequence"], 42) diff --git a/tools/management/report.py b/tools/management/report.py index 321a05d12..78b66dd52 100644 --- a/tools/management/report.py +++ b/tools/management/report.py @@ -15,6 +15,31 @@ from pathlib import Path HEADER, ENTRY, PER_PAGE, TAG, MAX_KEYS = 83, 13, 6, 16, 36 CURRENT_HEADER, CURRENT_PER_PAGE = 111, 4 +MAX_JSON_DEPTH = 64 + + +def _load_input_json(text): + # Do not rely on the interpreter's JSON scanner using recursive calls. + # Check before decoding so excessive containers never become a deep tree. + depth = 0 + quoted = escaped = False + for char in text: + if quoted: + if escaped: + escaped = False + elif char == "\\": + escaped = True + elif char == '"': + quoted = False + elif char == '"': + quoted = True + elif char in "[{": + depth += 1 + if depth > MAX_JSON_DEPTH: + raise ValueError("input JSON is nested too deeply") + elif char in "]}": + depth -= 1 + return json.loads(text) def _layout(payload): @@ -258,9 +283,10 @@ def main(): raise ValueError("administrator key must be 32 bytes") text = args.pages.read_text(encoding="utf-8") if args.mqtt: - messages = json.loads(text) if text.lstrip().startswith("[") else [json.loads(line) for line in text.splitlines() if line.strip()] + messages = (_load_input_json(text) if text.lstrip().startswith("[") else + [_load_input_json(line) for line in text.splitlines() if line.strip()]) else: - payloads = [bytes.fromhex(p) for p in json.loads(text)] + payloads = [bytes.fromhex(p) for p in _load_input_json(text)] password = getpass.getpass("Management password: ") if args.mqtt: reports = mqtt_reports(messages, password)