diff --git a/app.py b/app.py index 231d6a0..3e382fd 100644 --- a/app.py +++ b/app.py @@ -336,6 +336,14 @@ def parse_settings_form(post_data: str) -> Dict[str, list]: return parse_qs(post_data, keep_blank_values=True) +def parse_request_target(value: str): + """Parse an HTTP request target, returning None for malformed absolute URLs.""" + try: + return urlparse(value) + except ValueError: + return None + + def display_status_board_enabled(display: Dict[str, Any]) -> bool: """Return the status-board display flag, accepting the legacy key.""" if "show_status_board" in display: @@ -2469,6 +2477,13 @@ def send_json(self, status_code: int, payload: Dict[str, Any]): self.end_headers() self.wfile.write(json.dumps(payload, indent=2, ensure_ascii=False).encode("utf-8")) + def send_bad_request(self, message: str = "请求路径格式错误"): + self.send_response(400) + self.send_header("Content-Type", "text/plain; charset=utf-8") + self.send_no_cache_headers() + self.end_headers() + self.wfile.write(message.encode("utf-8")) + def is_api_write_authorized(self, parsed_path) -> bool: expected = configured_api_token() if not expected: @@ -2522,7 +2537,10 @@ def send_preference_cookie(self, name: str, value: str): ) def do_GET(self): - parsed_path = urlparse(self.path) + parsed_path = parse_request_target(self.path) + if parsed_path is None: + self.send_bad_request() + return path = parsed_path.path if path == "/" or path == "/index.html": @@ -2638,7 +2656,10 @@ def do_GET(self): self.wfile.write(b"

404 Not Found

") def do_POST(self): - parsed_path = urlparse(self.path) + parsed_path = parse_request_target(self.path) + if parsed_path is None: + self.send_bad_request() + return path = parsed_path.path if path == "/settings": diff --git a/tests/test_vibe_status.py b/tests/test_vibe_status.py index 4142c45..76effe2 100644 --- a/tests/test_vibe_status.py +++ b/tests/test_vibe_status.py @@ -418,6 +418,13 @@ def test_parse_settings_form_keeps_blank_host_for_normalization(self): self.assertEqual(updated["server"]["port"], 65535) self.assertEqual(updated["server"]["host"], "0.0.0.0") + def test_parse_request_target_rejects_malformed_absolute_url(self): + parsed = app.parse_request_target("/api/status?token=secret") + self.assertIsNotNone(parsed) + self.assertEqual(parsed.path, "/api/status") + + self.assertIsNone(app.parse_request_target("http://[::1")) + def test_write_json_atomic_preserves_existing_file_on_failure(self): target = Path(self.tmpdir.name) / "config.json" target.write_text('{"ok": true}', encoding="utf-8")