From 0e820fbf8be62652ce50d6954eb50edc5c7f0a33 Mon Sep 17 00:00:00 2001 From: John Hollowell Date: Thu, 4 Apr 2024 03:39:15 +0000 Subject: [PATCH 1/2] Fix improper spliting of non-exec QEMU commands Fixes: #161 --- proxmoxer/backends/https.py | 2 +- tests/api_mock.py | 47 +++++++++++++++++++++++++++++++++++++ tests/test_https.py | 38 +++++++++++++++++++++++++----- 3 files changed, 80 insertions(+), 7 deletions(-) diff --git a/proxmoxer/backends/https.py b/proxmoxer/backends/https.py index 3c210b8..3bb476a 100644 --- a/proxmoxer/backends/https.py +++ b/proxmoxer/backends/https.py @@ -194,7 +194,7 @@ def request( total_file_size = 0 for k, v in data.copy().items(): # split qemu exec commands for proper parsing by PVE (issue#89) - if k == "command": + if k == "command" and url.endswith("agent/exec"): if isinstance(v, list): data[k] = v elif "Windows" not in platform.platform(): diff --git a/tests/api_mock.py b/tests/api_mock.py index e0ae404..2e31a5a 100644 --- a/tests/api_mock.py +++ b/tests/api_mock.py @@ -100,6 +100,22 @@ def _generate_dynamic_responses(self): ) ) + resps.append( + responses.CallbackResponse( + method="GET", + url=re.compile(self.base_url + r"/nodes/[^/]+/qemu/[^/]+/agent/exec"), + callback=self._cb_echo, + ) + ) + + resps.append( + responses.CallbackResponse( + method="GET", + url=re.compile(self.base_url + r"/nodes/[^/]+/qemu/[^/]+/monitor"), + callback=self._cb_qemu_monitor, + ) + ) + resps.append( responses.CallbackResponse( method="GET", @@ -313,3 +329,34 @@ def _cb_url_metadata(self, request): } ), ) + + def _cb_qemu_monitor(self, request): + body = request.body + if body is not None: + if isinstance(body, MultipartEncoder): + body = body.to_string() # really, to byte string + body = body if isinstance(body, str) else str(body, "utf-8") + + # if the command is an array, throw the type error PVE would throw + if "&" in body: + return ( + 400, + self.common_headers, + json.dumps( + { + "data": None, + "errors": {"command": "type check ('string') failed - got ARRAY"}, + } + ), + ) + else: + resp = { + "method": request.method, + "url": request.url, + "headers": dict(request.headers), + "cookies": request._cookies.get_dict(), + "body": body, + # "body_json": dict(parse_qsl(request.body)), + } + print(resp) + return (200, self.common_headers, json.dumps(resp)) diff --git a/tests/test_https.py b/tests/test_https.py index da79304..53c3bcb 100644 --- a/tests/test_https.py +++ b/tests/test_https.py @@ -291,25 +291,51 @@ def test_request_data(self, mock_pve): assert content["body"] == "key=value" assert content["headers"]["Content-Type"] == "application/x-www-form-urlencoded" - def test_request_command_list(self, mock_pve): + def test_request_monitor_command_list(self, mock_pve): resp = self._session.request( - "GET", self.base_url + "/fake/echo", data={"command": ["echo", "hello", "world"]} + "GET", + self.base_url + "/nodes/node_name/qemu/100/monitor", + data={"command": ["info", "block"]}, + ) + + assert resp.status_code == 400 + + def test_request_exec_command_list(self, mock_pve): + resp = self._session.request( + "GET", + self.base_url + "/nodes/node_name/qemu/100/agent/exec", + data={"command": ["echo", "hello", "world"]}, ) content = resp.json() assert content["method"] == "GET" - assert content["url"] == self.base_url + "/fake/echo" + assert content["url"] == self.base_url + "/nodes/node_name/qemu/100/agent/exec" assert content["body"] == "command=echo&command=hello&command=world" assert content["headers"]["Content-Type"] == "application/x-www-form-urlencoded" - def test_request_command_string(self, mock_pve): + def test_request_monitor_command_string(self, mock_pve): resp = self._session.request( - "GET", self.base_url + "/fake/echo", data={"command": "echo hello world"} + "GET", + self.base_url + "/nodes/node_name/qemu/100/monitor", + data={"command": "echo hello world"}, ) content = resp.json() assert content["method"] == "GET" - assert content["url"] == self.base_url + "/fake/echo" + assert content["url"] == self.base_url + "/nodes/node_name/qemu/100/monitor" + assert content["body"] == "command=echo+hello+world" + assert content["headers"]["Content-Type"] == "application/x-www-form-urlencoded" + + def test_request_exec_command_string(self, mock_pve): + resp = self._session.request( + "GET", + self.base_url + "/nodes/node_name/qemu/100/agent/exec", + data={"command": "echo hello world"}, + ) + content = resp.json() + + assert content["method"] == "GET" + assert content["url"] == self.base_url + "/nodes/node_name/qemu/100/agent/exec" assert content["body"] == "command=echo&command=hello&command=world" assert content["headers"]["Content-Type"] == "application/x-www-form-urlencoded" From 76a9590ca7c81d419d09ba3aa1fc33077781a918 Mon Sep 17 00:00:00 2001 From: John Hollowell Date: Fri, 12 Apr 2024 00:13:28 +0000 Subject: [PATCH 2/2] Remove unneeded section in test mock --- tests/api_mock.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/tests/api_mock.py b/tests/api_mock.py index 2e31a5a..0488545 100644 --- a/tests/api_mock.py +++ b/tests/api_mock.py @@ -333,8 +333,6 @@ def _cb_url_metadata(self, request): def _cb_qemu_monitor(self, request): body = request.body if body is not None: - if isinstance(body, MultipartEncoder): - body = body.to_string() # really, to byte string body = body if isinstance(body, str) else str(body, "utf-8") # if the command is an array, throw the type error PVE would throw