diff --git a/py/api/logging_routes.py b/py/api/logging_routes.py index 77dc261..5329a1f 100644 --- a/py/api/logging_routes.py +++ b/py/api/logging_routes.py @@ -19,6 +19,16 @@ LOG_CONFIG_KEYS = frozenset( _LOG_INT_KEYS = frozenset({"max_file_size", "backup_count", "buffer_size"}) _LOG_BOOL_KEYS = frozenset({"console_logging", "file_logging"}) _LOG_LEVEL_NAMES = frozenset({"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}) +# Inclusive bounds for the integer keys that size the log files on disk +_MEGABYTE = 1024 * 1024 +_LOG_INT_BOUNDS = { + "max_file_size": (1 * _MEGABYTE, 1024 * _MEGABYTE), + "backup_count": (0, 50), +} +_LOG_INT_BOUND_LABELS = { + "max_file_size": "between 1 MB and 1024 MB (in bytes)", + "backup_count": "between 0 and 50", +} DEFAULT_LOG_LIMIT = 100 MAX_LOG_LIMIT = 500 @@ -64,6 +74,9 @@ def _validate_log_config(data): elif key in _LOG_INT_KEYS: if isinstance(value, bool) or not isinstance(value, int) or value < 0: return None, f"{key} must be a non-negative integer" + bounds = _LOG_INT_BOUNDS.get(key) + if bounds and not bounds[0] <= value <= bounds[1]: + return None, f"{key} must be {_LOG_INT_BOUND_LABELS[key]}" clean[key] = value elif key in _LOG_BOOL_KEYS: if not isinstance(value, bool): diff --git a/tests/test_api_logging.py b/tests/test_api_logging.py index 3052c6e..cda66ed 100644 --- a/tests/test_api_logging.py +++ b/tests/test_api_logging.py @@ -116,7 +116,7 @@ class LoggingAPITestCase(AioHTTPTestCase): sys.modules.pop("folder_paths", None) else: sys.modules["folder_paths"] = self._saved_folder_paths - self.api.db.close() + self.api.db.close_all() shutil.rmtree(self.tmpdir, ignore_errors=True) async def _post_config(self, payload): @@ -178,7 +178,7 @@ class TestUpdateLogConfigWhitelist(LoggingAPITestCase): async def test_valid_payload_is_applied(self): payload = { "level": "WARNING", - "max_file_size": 2048, + "max_file_size": 2 * 1024 * 1024, "backup_count": 0, "console_logging": False, "file_logging": True, @@ -192,6 +192,31 @@ class TestUpdateLogConfigWhitelist(LoggingAPITestCase): self.assertEqual(self.manager.update_calls, [payload]) self.assertEqual(data["config"]["level"], "WARNING") + async def test_max_file_size_is_bounded_to_1_to_1024_mb(self): + mb = 1024 * 1024 + for bad in (0, mb - 1, 1024 * mb + 1, 10**12): + resp = await self._post_config({"max_file_size": bad}) + self.assertEqual(resp.status, 400, bad) + self.assertIn("max_file_size", (await resp.json())["error"]) + self.assertEqual(self.manager.update_calls, []) + for ok in (mb, 1024 * mb): + resp = await self._post_config({"max_file_size": ok}) + self.assertEqual(resp.status, 200, ok) + self.assertEqual( + self.manager.update_calls, + [{"max_file_size": mb}, {"max_file_size": 1024 * mb}], + ) + + async def test_backup_count_is_bounded_to_0_to_50(self): + for bad in (-1, 51, 10**6): + resp = await self._post_config({"backup_count": bad}) + self.assertEqual(resp.status, 400, bad) + self.assertIn("backup_count", (await resp.json())["error"]) + self.assertEqual(self.manager.update_calls, []) + for ok in (0, 50): + resp = await self._post_config({"backup_count": ok}) + self.assertEqual(resp.status, 200, ok) + async def test_body_must_be_an_object(self): resp = await self._post_config(["level", "DEBUG"]) self.assertEqual(resp.status, 400)