diff --git a/py/api/logging_routes.py b/py/api/logging_routes.py index 9121a86..9c05298 100644 --- a/py/api/logging_routes.py +++ b/py/api/logging_routes.py @@ -4,6 +4,58 @@ import os from aiohttp import web +# Keys PromptManagerLogger.update_config() understands. Anything else is +# rejected so a client cannot inject arbitrary entries into the logging config. +LOG_CONFIG_KEYS = frozenset( + { + "level", + "max_file_size", + "backup_count", + "console_logging", + "file_logging", + "buffer_size", + } +) +_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"}) + + +def _validate_log_config(data): + """Validate a log-config payload against LOG_CONFIG_KEYS. + + Returns: + (clean, error): ``clean`` is the normalised dict to apply, ``error`` + a message describing the first problem (``clean`` is then ``None``). + """ + if not isinstance(data, dict): + return None, "Request body must be a JSON object" + + unknown = sorted(key for key in data if key not in LOG_CONFIG_KEYS) + if unknown: + return None, ( + f"Unknown logging config keys: {', '.join(unknown)}. " + f"Allowed: {', '.join(sorted(LOG_CONFIG_KEYS))}" + ) + + clean = {} + for key, value in data.items(): + if key == "level": + if not isinstance(value, str) or value.upper() not in _LOG_LEVEL_NAMES: + return None, ( + f"Invalid log level. Must be one of: {sorted(_LOG_LEVEL_NAMES)}" + ) + clean[key] = value.upper() + 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" + clean[key] = value + elif key in _LOG_BOOL_KEYS: + if not isinstance(value, bool): + return None, f"{key} must be a boolean" + clean[key] = value + return clean, None + class LoggingRoutesMixin: """Mixin providing logging-related API endpoints.""" @@ -163,24 +215,22 @@ class LoggingRoutesMixin: return web.json_response({"success": False, "error": str(e)}, status=500) async def update_log_config(self, request): - """Update logging configuration.""" + """Update logging configuration (whitelisted keys only).""" try: - data = await request.json() + try: + data = await request.json() + except ValueError: + return web.json_response( + {"success": False, "error": "Request body must be valid JSON"}, + status=400, + ) + + clean, error = _validate_log_config(data) + if error: + return web.json_response({"success": False, "error": error}, status=400) + logger_manager = self._get_logger_manager() - - if "level" in data: - valid_levels = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] - if data["level"].upper() not in valid_levels: - return web.json_response( - { - "success": False, - "error": f"Invalid log level. Must be one of: {valid_levels}", - }, - status=400, - ) - data["level"] = data["level"].upper() - - logger_manager.update_config(data) + logger_manager.update_config(clean) return web.json_response( { @@ -192,7 +242,10 @@ class LoggingRoutesMixin: except Exception as e: self.logger.error(f"Update log config error: {e}") - return web.json_response({"success": False, "error": str(e)}, status=500) + return web.json_response( + {"success": False, "error": "Failed to update logging configuration"}, + status=500, + ) async def get_log_stats(self, request): """Get logging statistics.""" diff --git a/tests/test_api_logging.py b/tests/test_api_logging.py new file mode 100644 index 0000000..12938bb --- /dev/null +++ b/tests/test_api_logging.py @@ -0,0 +1,196 @@ +"""Route tests for the logging API (py/api/logging_routes.py).""" + +import os +import shutil +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import MagicMock + +_mock_server = MagicMock() +_mock_server.PromptServer.instance.routes = MagicMock() +sys.modules["server"] = _mock_server + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from aiohttp import web # noqa: E402 +from aiohttp.test_utils import AioHTTPTestCase # noqa: E402 + +from database.operations import PromptDatabase # noqa: E402 +from py.api import PromptManagerAPI # noqa: E402 +from py.api.logging_routes import LOG_CONFIG_KEYS # noqa: E402 + + +class FakeLoggerManager: + """Stand-in for PromptManagerLogger with a scratch log directory.""" + + def __init__(self, log_dir): + self.log_dir = Path(log_dir) + self.config = { + "level": "INFO", + "max_file_size": 10 * 1024 * 1024, + "backup_count": 5, + "console_logging": True, + "file_logging": True, + "buffer_size": 1000, + } + self.update_calls = [] + self.logs = [] + + def update_config(self, new_config): + self.update_calls.append(dict(new_config)) + self.config.update(new_config) + + def get_config(self): + return dict(self.config) + + def get_recent_logs(self, limit=100, level=None): + logs = self.logs + if level: + logs = [entry for entry in logs if entry["level"] == level.upper()] + return list(reversed(logs[-limit:])) + + def get_log_files(self): + files = [] + for path in sorted(self.log_dir.glob("*.log*")): + files.append( + { + "filename": path.name, + "path": str(path), + "size": path.stat().st_size, + "modified": "2026-10-07T00:00:00", + "is_main": path.name == "prompt_manager.log", + } + ) + return files + + def truncate_logs(self): + truncated = [] + for path in self.log_dir.glob("prompt_manager.log*"): + path.write_text("") + truncated.append(path.name) + return {"truncated": truncated, "errors": []} + + def get_log_stats(self): + return { + "buffer_count": len(self.logs), + "level_counts": {}, + "log_files_count": len(self.get_log_files()), + "total_log_size": 0, + "log_directory": str(self.log_dir), + "current_level": self.config["level"], + } + + +class LoggingAPITestCase(AioHTTPTestCase): + + async def get_application(self): + self.tmpdir = tempfile.mkdtemp() + self.log_dir = Path(self.tmpdir) / "logs" + self.log_dir.mkdir() + self.manager = FakeLoggerManager(self.log_dir) + + app = web.Application() + routes = web.RouteTableDef() + self.api = PromptManagerAPI() + self.api.db = PromptDatabase(os.path.join(self.tmpdir, "prompts.db")) + self.api._get_logger_manager = lambda: self.manager + self.api.add_routes(routes) + app.router.add_routes(routes) + return app + + async def tearDownAsync(self): + shutil.rmtree(self.tmpdir, ignore_errors=True) + + async def _post_config(self, payload): + return await self.client.request( + "POST", "/prompt_manager/logs/config", json=payload + ) + + +class TestUpdateLogConfigWhitelist(LoggingAPITestCase): + + def test_whitelist_matches_logger_manager_keys(self): + self.assertEqual(LOG_CONFIG_KEYS, frozenset(self.manager.config.keys())) + + async def test_unknown_keys_rejected_and_listed(self): + resp = await self._post_config({"level": "INFO", "evil": 1, "log_dir": "/x"}) + + self.assertEqual(resp.status, 400) + data = await resp.json() + self.assertFalse(data["success"]) + self.assertIn("evil", data["error"]) + self.assertIn("log_dir", data["error"]) + self.assertEqual(self.manager.update_calls, []) + + async def test_invalid_level_rejected(self): + resp = await self._post_config({"level": "VERBOSE"}) + self.assertEqual(resp.status, 400) + self.assertEqual(self.manager.update_calls, []) + + async def test_level_must_be_a_string(self): + resp = await self._post_config({"level": 10}) + self.assertEqual(resp.status, 400) + self.assertEqual(self.manager.update_calls, []) + + async def test_level_is_normalised_to_upper_case(self): + resp = await self._post_config({"level": "debug"}) + self.assertEqual(resp.status, 200) + self.assertEqual(self.manager.update_calls, [{"level": "DEBUG"}]) + + async def test_negative_int_rejected(self): + resp = await self._post_config({"backup_count": -1}) + self.assertEqual(resp.status, 400) + self.assertEqual(self.manager.update_calls, []) + + async def test_numeric_string_rejected(self): + resp = await self._post_config({"max_file_size": "1024"}) + self.assertEqual(resp.status, 400) + self.assertEqual(self.manager.update_calls, []) + + async def test_bool_is_not_an_int(self): + resp = await self._post_config({"buffer_size": True}) + self.assertEqual(resp.status, 400) + self.assertEqual(self.manager.update_calls, []) + + async def test_flags_must_be_bool(self): + resp = await self._post_config({"console_logging": "yes"}) + self.assertEqual(resp.status, 400) + self.assertEqual(self.manager.update_calls, []) + + async def test_valid_payload_is_applied(self): + payload = { + "level": "WARNING", + "max_file_size": 2048, + "backup_count": 0, + "console_logging": False, + "file_logging": True, + "buffer_size": 50, + } + resp = await self._post_config(payload) + + self.assertEqual(resp.status, 200) + data = await resp.json() + self.assertTrue(data["success"]) + self.assertEqual(self.manager.update_calls, [payload]) + self.assertEqual(data["config"]["level"], "WARNING") + + async def test_body_must_be_an_object(self): + resp = await self._post_config(["level", "DEBUG"]) + self.assertEqual(resp.status, 400) + self.assertEqual(self.manager.update_calls, []) + + async def test_malformed_json_is_400(self): + resp = await self.client.request( + "POST", + "/prompt_manager/logs/config", + data=b"{not json", + headers={"Content-Type": "application/json"}, + ) + self.assertEqual(resp.status, 400) + self.assertEqual(self.manager.update_calls, []) + + +if __name__ == "__main__": + unittest.main()