feat: harden training API validation

src/api/server.py

@@ -40,6 +40,7 @@ def build_api_dependencies() -> dict:
 
 
 STATIC_DIR = Path(__file__).resolve().parents[1] / "web"
+MAX_TRAINING_LIMIT = 100
 
 
 def route_request(
@@ -80,7 +81,7 @@ def route_request(
         status = _get_single(query, "status")
         skill_tag = _get_single(query, "skill_tag")
         instrument_id = _get_optional_int(query, "instrument_id")
-        limit = _get_int(query, "limit", 100)
+        limit = _get_limit(query, "limit", 100, max_value=MAX_TRAINING_LIMIT)
         return 200, {
             "items": deps["training_cases"](
                 status=status,
@@ -106,7 +107,7 @@ def route_request(
     if method == "GET" and path == "/training/progress":
         user_id = _get_single(query, "user_id", "local")
         skill_tag = _get_single(query, "skill_tag")
-        limit = _get_int(query, "limit", 100)
+        limit = _get_limit(query, "limit", 100, max_value=MAX_TRAINING_LIMIT)
         return 200, {"items": deps["training_progress"](user_id=user_id, skill_tag=skill_tag, limit=limit)}
     return 404, {"error": "not_found"}
 
@@ -120,11 +121,11 @@ class APIRequestHandler(BaseHTTPRequestHandler):
             self._serve_static(parsed.path)
             return
         try:
-            status, payload = route_request(parsed.path, parse_qs(parsed.query), dependencies=self.dependencies)
+            status, payload = route_request(parsed.path, parse_qs(parsed.query, keep_blank_values=True), dependencies=self.dependencies)
         except ValueError as exc:
             status, payload = 400, {"error": str(exc)}
-        except Exception as exc:
-            status, payload = 500, {"error": "internal_server_error", "detail": str(exc)}
+        except Exception:
+            status, payload = 500, {"error": "internal_server_error"}
         self._send_json(status, payload)
 
     def do_POST(self) -> None:
@@ -140,8 +141,8 @@ class APIRequestHandler(BaseHTTPRequestHandler):
             )
         except ValueError as exc:
             status, payload = 400, {"error": str(exc)}
-        except Exception as exc:
-            status, payload = 500, {"error": "internal_server_error", "detail": str(exc)}
+        except Exception:
+            status, payload = 500, {"error": "internal_server_error"}
         self._send_json(status, payload)
 
     def _send_json(self, status: int, payload: dict) -> None:
@@ -206,6 +207,19 @@ def _get_int(query: dict, key: str, default: int) -> int:
         raise ValueError(f"invalid_int_query_param:{key}") from exc
 
 
+def _get_limit(query: dict, key: str, default: int, max_value: int) -> int:
+    value = _get_single(query, key)
+    if value is None:
+        return default
+    try:
+        parsed = int(value)
+    except ValueError as exc:
+        raise ValueError(f"invalid_limit_query_param:{key}") from exc
+    if parsed < 1 or parsed > max_value:
+        raise ValueError(f"invalid_limit_query_param:{key}")
+    return parsed
+
+
 def _get_single(query: dict, key: str, default: Optional[str] = None) -> Optional[str]:
     values = query.get(key)
     if not values:

tests/unit/test_training_api_m4.py

@@ -0,0 +1,127 @@
+import io
+import json
+import unittest
+
+from src.api.server import APIRequestHandler
+
+
+def _dependencies(cases=None, progress=None) -> dict:
+    return {
+        "training_cases": cases or (lambda **kwargs: []),
+        "training_progress": progress or (lambda **kwargs: []),
+    }
+
+
+class _GetHandlerHarness(APIRequestHandler):
+    def __init__(self, path: str, dependencies: dict) -> None:
+        self.path = path
+        self.wfile = io.BytesIO()
+        self.dependencies = dependencies
+        self.status = None
+        self.response_headers = {}
+
+    def send_response(self, code, message=None):  # noqa: D401
+        self.status = code
+
+    def send_header(self, keyword, value):
+        self.response_headers[keyword] = value
+
+    def end_headers(self):
+        return None
+
+    def log_message(self, format, *args):
+        return None
+
+    def json_body(self):
+        return json.loads(self.wfile.getvalue().decode("utf-8"))
+
+
+class TrainingApiM4Tests(unittest.TestCase):
+    def test_training_cases_rejects_invalid_limits(self) -> None:
+        calls = []
+
+        def list_cases(**kwargs):
+            calls.append(kwargs)
+            return []
+
+        for path in (
+            "/training/cases?limit=",
+            "/training/cases?limit=0",
+            "/training/cases?limit=-1",
+            "/training/cases?limit=101",
+            "/training/cases?limit=not-a-number",
+        ):
+            with self.subTest(path=path):
+                handler = _GetHandlerHarness(path, _dependencies(cases=list_cases))
+
+                handler.do_GET()
+
+                self.assertEqual(handler.status, 400)
+                self.assertEqual(handler.json_body(), {"error": "invalid_limit_query_param:limit"})
+
+        self.assertEqual(calls, [])
+
+    def test_training_progress_rejects_invalid_limits(self) -> None:
+        calls = []
+
+        def get_progress(**kwargs):
+            calls.append(kwargs)
+            return []
+
+        for path in (
+            "/training/progress?limit=",
+            "/training/progress?limit=0",
+            "/training/progress?limit=-1",
+            "/training/progress?limit=101",
+            "/training/progress?limit=not-a-number",
+        ):
+            with self.subTest(path=path):
+                handler = _GetHandlerHarness(path, _dependencies(progress=get_progress))
+
+                handler.do_GET()
+
+                self.assertEqual(handler.status, 400)
+                self.assertEqual(handler.json_body(), {"error": "invalid_limit_query_param:limit"})
+
+        self.assertEqual(calls, [])
+
+    def test_training_limits_accept_upper_bound(self) -> None:
+        case_calls = []
+        progress_calls = []
+
+        def list_cases(**kwargs):
+            case_calls.append(kwargs)
+            return []
+
+        def get_progress(**kwargs):
+            progress_calls.append(kwargs)
+            return []
+
+        cases_handler = _GetHandlerHarness("/training/cases?limit=100", _dependencies(cases=list_cases))
+        progress_handler = _GetHandlerHarness("/training/progress?limit=100", _dependencies(progress=get_progress))
+
+        cases_handler.do_GET()
+        progress_handler.do_GET()
+
+        self.assertEqual(cases_handler.status, 200)
+        self.assertEqual(progress_handler.status, 200)
+        self.assertEqual(case_calls[0]["limit"], 100)
+        self.assertEqual(progress_calls[0]["limit"], 100)
+
+    def test_unknown_exception_response_hides_internal_detail(self) -> None:
+        def list_cases(**kwargs):
+            raise RuntimeError("raw internal failure marker")
+
+        handler = _GetHandlerHarness("/training/cases", _dependencies(cases=list_cases))
+
+        handler.do_GET()
+
+        body = handler.json_body()
+        serialized = json.dumps(body)
+        self.assertEqual(handler.status, 500)
+        self.assertEqual(body, {"error": "internal_server_error"})
+        self.assertNotIn("raw internal failure marker", serialized)
+
+
+if __name__ == "__main__":
+    unittest.main()