From 3254ca6d0526a05466aa4c4a15cbe97e99a6aa27 Mon Sep 17 00:00:00 2001 From: VAIO73 <50487331+Vaso73@users.noreply.github.com> Date: Wed, 14 Oct 2026 17:02:41 +0000 Subject: [PATCH] fix: confirm VM control tasks safely --- AppImage/scripts/flask_server.py | 56 ++++++++++++++++--- AppImage/scripts/lxc_check_job.py | 16 ++++-- tests/lxc_updates/test_check_only_job.py | 9 ++- .../test_control_task_reservation.py | 32 +++++++++-- tests/lxc_updates/test_vm_control_task.py | 45 +++++++++++++-- 5 files changed, 132 insertions(+), 26 deletions(-) diff --git a/AppImage/scripts/flask_server.py b/AppImage/scripts/flask_server.py index d25dcadb..6a6ddf42 100644 --- a/AppImage/scripts/flask_server.py +++ b/AppImage/scripts/flask_server.py @@ -15876,6 +15876,7 @@ def api_vm_config_set(vmid): def api_vm_control(vmid): """Control VM/LXC (start, stop, shutdown, reboot)""" operation_lock = None + task_identity_confirmed = False try: data = request.get_json() action = data.get('action') # start, stop, shutdown, reboot @@ -15912,6 +15913,35 @@ def api_vm_control(vmid): capture_output=True, text=True, timeout=30) if control_result.returncode == 0: + import re + # pvesh can print progress lines before the task identity. + # The final non-empty line is the only task associated with + # this request; accepting a different log line would make the + # frontend poll an unrelated guest operation. + lines = [line.strip().strip('"') for line in + (control_result.stdout or '').splitlines() if line.strip()] + task = lines[-1] if lines else '' + task_types = [f"{'vz' if vm_type == 'lxc' else 'qm'}{action}"] + if action == 'start': + task_types.append('hastart') + elif action == 'stop': + task_types.append('hastop') + task_pattern = (r'UPID:[\w.-]+:[0-9A-Fa-f]+:[0-9A-Fa-f]+:' + r'[0-9A-Fa-f]+:(?:' + '|'.join(task_types) + + rf'):{vmid}:[\w.@!-]+:') + if not re.fullmatch(task_pattern, task): + # A successful process exit without a verifiable task + # cannot be polled. Do not leave a task-less control + # marker that would block this CT forever. + if operation_lock is not None: + lxc_check_job.cancel_control(operation_lock) + return jsonify({ + 'success': False, + 'vmid': vmid, + 'action': action, + 'error': 'Control task was not confirmed by Proxmox', + }), 502 + task_identity_confirmed = True # Invalidate VM resources cache so the next /api/vms call # returns fresh status instead of the pre-action snapshot. _pvesh_cache['cluster_resources_vm_time'] = 0 @@ -15919,17 +15949,16 @@ def api_vm_control(vmid): # (lxc_ip_info, os_info) — a start/stop flips those, so # drop the modal caches for this vmid. _vm_cache_invalidate(vmid) - # Keep the legacy acceptance response; new clients also wait - # for this specific task instead of assuming it has finished. - task = (control_result.stdout or '').strip().strip('"') - if operation_lock is not None and task.startswith('UPID:'): + # The frontend waits for this exact task rather than assuming + # that accepting the request means the action completed. + if operation_lock is not None: lxc_check_job.retain_control(vmid, task, operation_lock) operation_lock = None return jsonify({ 'success': True, 'vmid': vmid, 'action': action, - 'task': task if task.startswith('UPID:') else None, + 'task': task, 'message': f'Successfully executed {action} on {vm_info.get("name")}' }) else: @@ -15986,6 +16015,14 @@ def api_vm_control(vmid): else: return jsonify({'error': 'Failed to get VM details'}), 500 except Exception as e: + # A transport error happens before we have an identity that can be + # recovered later. Leaving the pre-submission marker would otherwise + # make the CT permanently busy after one timed-out pvesh invocation. + if operation_lock is not None and not task_identity_confirmed: + try: + lxc_check_job.cancel_control(operation_lock) + except Exception: + pass return jsonify({'error': str(e)}), 500 finally: @@ -15999,7 +16036,8 @@ def api_vm_control_task(vmid, task): import re import json if not re.fullmatch(r'UPID:[\w.-]+:[0-9A-Fa-f]+:[0-9A-Fa-f]+:[0-9A-Fa-f]+:' - r'(?:vz|qm)(?:start|stop|shutdown|reboot):[0-9]+:[\w.@-]+:', task): + r'(?:(?:vz|qm)(?:start|stop|shutdown|reboot)|ha(?:start|stop)):' + r'[0-9]+:[\w.@!-]+:', task): return jsonify({'error': 'Invalid control task'}), 400 parts = task.split(':') resource = next((item for item in get_cached_pvesh_cluster_resources_vm() or [] @@ -16009,7 +16047,11 @@ def api_vm_control_task(vmid, task): vm_type = 'lxc' if resource.get('type') == 'lxc' else 'qemu' node = resource.get('node', 'pve') prefix = 'vz' if vm_type == 'lxc' else 'qm' - if parts[1] != node or parts[6] != str(vmid) or not parts[5].startswith(prefix): + allowed_task_types = {f'{prefix}{action}' for action in + ('start', 'stop', 'shutdown', 'reboot')} + allowed_task_types.update(('hastart', 'hastop')) + if (parts[1] != node or parts[6] != str(vmid) + or parts[5] not in allowed_task_types): return jsonify({'error': 'Control task does not belong to this guest'}), 400 try: result = subprocess.run(['pvesh', 'get', f'/nodes/{node}/tasks/{task}/status', diff --git a/AppImage/scripts/lxc_check_job.py b/AppImage/scripts/lxc_check_job.py index 91f288cc..0ca2d733 100644 --- a/AppImage/scripts/lxc_check_job.py +++ b/AppImage/scripts/lxc_check_job.py @@ -25,6 +25,7 @@ _jobs = {} _attempts = {} COOLDOWN = 60 JOB_TTL = 3600 +CONTROL_TASK_WAIT_SECONDS = 180 def acquire(vmid): @@ -104,7 +105,7 @@ def begin_control(handle, node): def cancel_control(handle): - """Only a definite submission failure can clear an unconfirmed request.""" + """Clear a request for which no recoverable task identity was recorded.""" _write_control(handle, None) @@ -129,7 +130,8 @@ def check_control_task(vmid, handle=None): raise ValueError('Invalid node') if (not isinstance(task, str) or not re.fullmatch(r'UPID:[\w.-]+:[0-9A-Fa-f]+:[0-9A-Fa-f]+:' - r'[0-9A-Fa-f]+:vz(?:start|stop|shutdown|reboot):' + r'[0-9A-Fa-f]+:(?:vz(?:start|stop|shutdown|reboot)' + r'|ha(?:start|stop)):' r'[0-9]+:[\w.@!-]+:', task) or task.split(':')[1] != node or task.split(':')[6] != str(vmid)): raise ValueError('Unconfirmed control task') @@ -146,14 +148,18 @@ def check_control_task(vmid, handle=None): def _wait_control(vmid, reservation): + deadline = time.monotonic() + CONTROL_TASK_WAIT_SECONDS try: - while True: + while time.monotonic() < deadline: try: check_control_task(vmid, reservation) return except BlockingIOError: - # A status timeout must not unlock a task that may still run. - time.sleep(2) + # Keep the known task record for recovery after this bounded + # waiter exits; a later caller asks Proxmox to verify it. + remaining = deadline - time.monotonic() + if remaining > 0: + time.sleep(min(2, remaining)) finally: reservation.close() diff --git a/tests/lxc_updates/test_check_only_job.py b/tests/lxc_updates/test_check_only_job.py index 3fa35698..2953ac57 100644 --- a/tests/lxc_updates/test_check_only_job.py +++ b/tests/lxc_updates/test_check_only_job.py @@ -290,8 +290,13 @@ class CheckOnlyTests(unittest.TestCase): def test_control_route_uses_same_lock_and_preserves_qemu(self): app = Flask(__name__) - command = Mock(return_value=subprocess.CompletedProcess( - [], 0, '"UPID:fixture:000A:000B:000C:vzstart:100:root@pam:"\n', '')) + def control_result(argv, **_): + path = argv[2] + vm_type, vmid, action = path.split('/')[3], path.split('/')[4], path.rsplit('/', 1)[1] + task_type = ('vz' if vm_type == 'lxc' else 'qm') + action + return subprocess.CompletedProcess( + [], 0, f'"UPID:fixture:000A:000B:000C:{task_type}:{vmid}:root@pam:"\n', '') + command = Mock(side_effect=control_result) scope = {'app': app, 'jsonify': jsonify, 'request': request, 'require_auth': lambda fn: fn, 'get_cached_pvesh_cluster_resources_vm': lambda: [ diff --git a/tests/lxc_updates/test_control_task_reservation.py b/tests/lxc_updates/test_control_task_reservation.py index 378d4b59..d09c5755 100644 --- a/tests/lxc_updates/test_control_task_reservation.py +++ b/tests/lxc_updates/test_control_task_reservation.py @@ -109,18 +109,38 @@ class ControlReservationTests(unittest.TestCase): job.acquire(100).close() probe.assert_not_called() - def test_definite_failure_clears_marker_but_timeout_or_missing_upid_cannot_confirm_completion(self): + def test_definite_failure_or_unidentified_submission_never_leaves_a_permanent_reservation(self): self.route.run.return_value = self.result({}, code=1) self.assertEqual(self.route.client.post('/api/vms/100/control', json={'action': 'start'}).status_code, 500) self.assertEqual(self.lock_path.read_text(), '') for response in [subprocess.TimeoutExpired('pvesh', 30), self.result('')]: self.route.run.side_effect = response if isinstance(response, Exception) else None self.route.run.return_value = response - self.route.client.post('/api/vms/100/control', json={'action': 'start'}) - self.assertIsNone(json.loads(self.lock_path.read_text())['task']) - with self.assertRaises(BlockingIOError): - job.acquire(100) - self.lock_path.write_text('') # fixture reset after explicit assertions + response = self.route.client.post('/api/vms/100/control', json={'action': 'start'}) + self.assertIn(response.status_code, (500, 502), response.json) + self.assertEqual(self.lock_path.read_text(), '') + job.acquire(100).close() + + def test_ha_task_recovers_like_a_native_lxc_task(self): + task = TASK.replace('vzreboot', 'hastop') + self.lock_path.write_text(json.dumps({'node': 'fixture', 'task': task})) + with patch.object(job.subprocess, 'run', return_value=self.result( + {'status': 'stopped', 'exitstatus': 'OK'})) as probe: + job.acquire(100).close() + self.assertEqual(probe.call_count, 1) + self.assertEqual(self.lock_path.read_text(), '') + + def test_wait_deadline_releases_fd_but_keeps_known_task_for_recovery(self): + handle = self.accepted() + with patch.object(job, 'CONTROL_TASK_WAIT_SECONDS', 0): + job._wait_control(100, handle) + self.assertTrue(handle.closed) + record = json.loads(self.lock_path.read_text()) + self.assertEqual(record['task'], TASK) + with patch.object(job.subprocess, 'run', return_value=self.result( + {'status': 'stopped', 'exitstatus': 'OK'})): + job.acquire(100).close() + self.assertEqual(self.lock_path.read_text(), '') def test_thread_launch_failure_retains_marker_for_recovery(self): self.thread.start.side_effect = RuntimeError('thread failed') diff --git a/tests/lxc_updates/test_vm_control_task.py b/tests/lxc_updates/test_vm_control_task.py index fcf4ff21..fdb6d1c6 100644 --- a/tests/lxc_updates/test_vm_control_task.py +++ b/tests/lxc_updates/test_vm_control_task.py @@ -57,12 +57,33 @@ class ControlTaskTests(unittest.TestCase): self.assertEqual(retained.call_count, 4) lock.close.assert_not_called() - def test_legacy_acceptance_without_task_and_invalid_action(self): - self.run.return_value = subprocess.CompletedProcess([], 0, '', '') - response = self.client.post('/api/vms/200/control', json={'action': 'start'}) - self.assertEqual(response.status_code, 200) - self.assertTrue(response.json['success']) - self.assertIsNone(response.json['task']) + def test_control_post_reads_final_upid_after_pvesh_log_output(self): + lock = Mock() + retained = Mock() + module = SimpleNamespace(acquire=lambda _: lock, begin_control=Mock(), + retain_control=retained, cancel_control=Mock()) + task = TASK.replace('vzreboot', 'vzstart') + self.run.return_value = subprocess.CompletedProcess( + [], 0, f'waiting for server lock\n"{task}"\n', '') + with patch.dict('sys.modules', {'lxc_check_job': module}): + response = self.client.post('/api/vms/100/control', json={'action': 'start'}) + self.assertEqual(response.status_code, 200, response.json) + self.assertEqual(response.json['task'], task) + retained.assert_called_once_with(100, task, lock) + + def test_control_post_rejects_success_without_a_confirmable_task(self): + lock = Mock() + module = SimpleNamespace(acquire=lambda _: lock, begin_control=Mock(), + retain_control=Mock(), cancel_control=Mock()) + self.run.return_value = subprocess.CompletedProcess([], 0, 'waiting for server lock\n', '') + with patch.dict('sys.modules', {'lxc_check_job': module}): + response = self.client.post('/api/vms/100/control', json={'action': 'start'}) + self.assertEqual(response.status_code, 502, response.json) + self.assertFalse(response.json['success']) + self.assertIn('task', response.json['error'].lower()) + module.retain_control.assert_not_called() + module.cancel_control.assert_called_once_with(lock) + lock.close.assert_called_once() self.run.reset_mock() self.assertEqual(self.client.post('/api/vms/100/control', json={'action': 'destroy'}).status_code, 400) self.run.assert_not_called() @@ -91,6 +112,18 @@ class ControlTaskTests(unittest.TestCase): self.assertEqual(self.run.call_count, 2) self.assertTrue(all(call.args[0][:2] == ['pvesh', 'get'] for call in self.run.call_args_list)) + def test_ha_start_and_stop_tasks_belong_to_the_guest(self): + for action, expected in [('hastart', 'running'), ('hastop', 'stopped')]: + with self.subTest(action=action): + self.run.reset_mock() + self.run.side_effect = [self.result(dict(status='stopped', exitstatus='OK')), + self.result(dict(status=expected))] + task = TASK.replace('vzreboot', action) + response = self.get(task) + self.assertEqual(response.status_code, 200, response.json) + self.assertEqual(response.json['guest_status'], expected) + self.assertEqual(self.run.call_count, 2) + def test_wrong_guest_node_type_and_path_make_zero_probes(self): for task in [TASK.replace(':100:', ':200:'), TASK.replace('fixture', 'other'), TASK.replace('vzreboot', 'qmreboot'), TASK.replace('vzreboot', 'vzdump'),